prehistoric-systems/omnivore
1
0
Fork 0
mirror of https://github.com/omnivore-app/omnivore.git synced 2026-03-11 08:54:26 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #3359 from omnivore-app/feat/backend-bull-job-processor

Browse source 
Use bullmq to process jobs
This commit is contained in:
Jackson Harper 2024-01-18 15:38:30 +08:00 committed by GitHub
commit bffe31bcc7
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
27 changed files with 2497 additions and 196 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
packages/api/.env.example
View file

@ -23,7 +23,6 @@ SAMPLE_METRICS_LOCALLY=FALSE
GCS_UPLOAD_BUCKET=
GCS_UPLOAD_SA_KEY_FILE_PATH=
TWITTER_BEARER_TOKEN=
PREVIEW_IMAGE_WRAPPER_ID='selected_highlight_wrapper'
SENDER_MESSAGE=msgs@sender.domain
SENDER_FEEDBACK=feedback@sender.domain
SENDER_GENERAL=no-reply@sender.domain

1
packages/api/.env.test
View file

@ -24,7 +24,6 @@ GCS_UPLOAD_BUCKET=
GCS_UPLOAD_SA_KEY_FILE_PATH=
GCS_UPLOAD_PRIVATE_BUCKET=
TWITTER_BEARER_TOKEN=
PREVIEW_IMAGE_WRAPPER_ID='selected_highlight_wrapper'
SEGMENT_WRITE_KEY='test'
PUBSUB_VERIFICATION_TOKEN='123456'
CONTENT_FETCH_URL=http://localhost:9090/

5
packages/api/package.json
View file

@ -5,7 +5,9 @@
"scripts": {
"build": "tsc && yarn copy-files",
"dev": "ts-node-dev --files src/server.ts",
"dev_qp": "ts-node-dev --files src/queue-processor.ts",
"start": "node dist/server.js",
"start_queue_processor": "node dist/queue-processor.js",
"lint": "eslint src --ext ts,js,tsx,jsx",
"lint:fix": "eslint src --fix --ext ts,js,tsx,jsx",
"test:typecheck": "tsc --noEmit",
@ -47,6 +49,7 @@
"apollo-server-express": "^3.6.3",
"axios": "^0.27.2",
"bcryptjs": "^2.4.3",
"bullmq": "^5.1.1",
"cookie": "^0.5.0",
"cookie-parser": "^1.4.5",
"cors": "^2.8.5",
@ -152,4 +155,4 @@
"volta": {
"extends": "../../package.json"
}
}
}

110
packages/api/src/jobs/rss/refreshAllFeeds.ts Normal file
View file

@ -0,0 +1,110 @@
import { Job, Queue } from 'bullmq'
import { DataSource } from 'typeorm'
import { QUEUE_NAME } from '../../queue-processor'
import { redisDataSource } from '../../redis_data_source'
import { RssSubscriptionGroup } from '../../utils/createTask'
import { stringToHash } from '../../utils/helpers'
export const refreshAllFeeds = async (db: DataSource): Promise<boolean> => {
const subscriptionGroups = (await db.createEntityManager().query(
`
SELECT
url,
ARRAY_AGG(id) AS "subscriptionIds",
ARRAY_AGG(user_id) AS "userIds",
ARRAY_AGG(last_fetched_at) AS "fetchedDates",
ARRAY_AGG(coalesce(scheduled_at, NOW())) AS "scheduledDates",
ARRAY_AGG(last_fetched_checksum) AS checksums,
ARRAY_AGG(fetch_content) AS "fetchContents",
ARRAY_AGG(coalesce(folder, $3)) AS folders
FROM
omnivore.subscriptions
WHERE
type = $1
AND status = $2
AND (scheduled_at <= NOW() OR scheduled_at IS NULL)
GROUP BY
url
`,
['RSS', 'ACTIVE', 'following']
)) as RssSubscriptionGroup[]
for (const group of subscriptionGroups) {
try {
await updateSubscriptionGroup(group)
} catch (err) {
// we don't want to fail the whole job if one subscription group fails
console.error('error updating subscription group')
}
}
return true
}
const updateSubscriptionGroup = async (group: RssSubscriptionGroup) => {
const feedURL = group.url
const userList = JSON.stringify(group.userIds.sort())
if (!feedURL) {
console.error('no url for feed group', group)
return
}
if (!userList) {
console.error('no userlist for feed group', group)
return
}
const jobid = `refresh-feed_${stringToHash(feedURL)}_${stringToHash(
userList
)}`
const payload = {
subscriptionIds: group.subscriptionIds,
feedUrl: group.url,
lastFetchedTimestamps: group.fetchedDates.map(
(timestamp) => timestamp?.getTime() || 0
), // unix timestamp in milliseconds
lastFetchedChecksums: group.checksums,
scheduledTimestamps: group.scheduledDates.map((timestamp) =>
timestamp.getTime()
), // unix timestamp in milliseconds
userIds: group.userIds,
fetchContents: group.fetchContents,
folders: group.folders,
}
await queueRSSRefreshFeedJob(jobid, payload)
}
const createBackendQueue = (): Queue | undefined => {
if (!redisDataSource.workerRedisClient) {
throw new Error('Can not create queues, redis is not initialized')
}
return new Queue(QUEUE_NAME, {
connection: redisDataSource.workerRedisClient,
})
}
export const queueRSSRefreshAllFeedsJob = async () => {
const queue = createBackendQueue()
if (!queue) {
return false
}
return queue.add('refresh-all-feeds', {})
}
type QueuePriority = 'low' | 'high'
export const queueRSSRefreshFeedJob = async (
jobid: string,
payload: any,
options = { priority: 'high' as QueuePriority }
): Promise<Job | undefined> => {
const queue = createBackendQueue()
if (!queue) {
return undefined
}
return queue.add('refresh-feed', payload, {
jobId: jobid,
removeOnComplete: true,
removeOnFail: true,
lifo: options.priority == 'high',
})
}

664
packages/api/src/jobs/rss/refreshFeed.ts Normal file
View file

@ -0,0 +1,664 @@
import axios from 'axios'
import crypto from 'crypto'
import * as dotenv from 'dotenv' // see https://github.com/motdotla/dotenv#how-do-i-use-dotenv-with-import
import * as jwt from 'jsonwebtoken'
import { parseHTML } from 'linkedom'
import Parser, { Item } from 'rss-parser'
import { promisify } from 'util'
import createHttpTaskWithToken from '../../utils/createTask'
import { env } from '../../env'
import { redisDataSource } from '../../redis_data_source'
type FolderType = 'following' | 'inbox'
interface RefreshFeedRequest {
subscriptionIds: string[]
feedUrl: string
lastFetchedTimestamps: number[] // unix timestamp in milliseconds
scheduledTimestamps: number[] // unix timestamp in milliseconds
lastFetchedChecksums: string[]
userIds: string[]
fetchContents: boolean[]
folders: FolderType[]
}
export const isRefreshFeedRequest = (data: any): data is RefreshFeedRequest => {
return (
'subscriptionIds' in data &&
'feedUrl' in data &&
'lastFetchedTimestamps' in data &&
'scheduledTimestamps' in data &&
'userIds' in data &&
'lastFetchedChecksums' in data &&
'fetchContents' in data &&
'folders' in data
)
}
// link can be a string or an object
type RssFeedItemLink = string | { $: { rel?: string; href: string } }
type RssFeed = Parser.Output<{
published?: string
updated?: string
created?: string
link?: RssFeedItemLink
links?: RssFeedItemLink[]
}> & {
lastBuildDate?: string
'syn:updatePeriod'?: string
'syn:updateFrequency'?: string
'sy:updatePeriod'?: string
'sy:updateFrequency'?: string
}
type RssFeedItemMedia = {
$: { url: string; width?: string; height?: string; medium?: string }
}
export type RssFeedItem = Item & {
'media:thumbnail'?: RssFeedItemMedia
'media:content'?: RssFeedItemMedia[]
link: string
}
export const isOldItem = (item: RssFeedItem, lastFetchedAt: number) => {
// existing items and items that were published before 24h
const publishedAt = item.isoDate ? new Date(item.isoDate) : new Date()
return (
publishedAt <= new Date(lastFetchedAt) ||
publishedAt < new Date(Date.now() - 24 * 60 * 60 * 1000)
)
}
const feedFetchFailedRedisKey = (feedUrl: string) =>
`feed-fetch-failure:${feedUrl}`
const isFeedBlocked = async (feedUrl: string) => {
const key = feedFetchFailedRedisKey(feedUrl)
const redisClient = redisDataSource.redisClient
try {
const result = await redisClient?.get(key)
// if the feed has failed to fetch more than certain times, block it
const maxFailures = parseInt(process.env.MAX_FEED_FETCH_FAILURES ?? '10')
if (result && parseInt(result) > maxFailures) {
console.log('feed is blocked: ', feedUrl)
return true
}
} catch (error) {
console.error('Failed to check feed block status', feedUrl, error)
}
return false
}
const incrementFeedFailure = async (feedUrl: string) => {
const redisClient = redisDataSource.redisClient
const key = feedFetchFailedRedisKey(feedUrl)
try {
const result = await redisClient?.incr(key)
// expire the key in 1 day
await redisClient?.expire(key, 24 * 60 * 60)
return result
} catch (error) {
console.error('Failed to block feed', feedUrl, error)
return null
}
}
export const isContentFetchBlocked = (feedUrl: string) => {
if (feedUrl.startsWith('https://arxiv.org/')) {
return true
}
if (feedUrl.startsWith('https://lwn.net/headlines/newrss')) {
return true
}
return false
}
const getThumbnail = (item: RssFeedItem) => {
if (item['media:thumbnail']) {
return item['media:thumbnail'].$.url
}
return item['media:content']?.find((media) => media.$.medium === 'image')?.$
.url
}
export const fetchAndChecksum = async (url: string) => {
try {
const response = await axios.get(url, {
responseType: 'arraybuffer',
timeout: 60_000,
maxRedirects: 10,
headers: {
'User-Agent':
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.0.0 Safari/537.36',
Accept:
'application/rss+xml, application/rdf+xml;q=0.8, application/atom+xml;q=0.6, application/xml;q=0.4, text/xml, text/html;q=0.4',
},
})
const hash = crypto.createHash('sha256')
hash.update(response.data as Buffer)
const dataStr = (response.data as Buffer).toString()
return { url, content: dataStr, checksum: hash.digest('hex') }
} catch (error) {
console.log(`Failed to fetch or hash content from ${url}.`, error)
return null
}
}
const parseFeed = async (url: string, content: string) => {
try {
// check if url is a telegram channel
const telegramRegex = /https:\/\/t\.me\/([a-zA-Z0-9_]+)/
const telegramMatch = url.match(telegramRegex)
if (telegramMatch) {
const dom = parseHTML(content).document
const title = dom.querySelector('meta[property="og:title"]')
// post has attribute data-post
const posts = dom.querySelectorAll('[data-post]')
const items = Array.from(posts)
.map((post) => {
const id = post.getAttribute('data-post')
if (!id) {
return null
}
const url = `https://t.me/${telegramMatch[1]}/${id}`
// find the <time> element
const time = post.querySelector('time')
const dateTime = time?.getAttribute('datetime') || undefined
return {
link: url,
isoDate: dateTime,
}
})
.filter((item) => !!item) as RssFeedItem[]
return {
title: title?.getAttribute('content') || dom.title,
items,
}
}
// return await is needed to catch errors thrown by the parser
// otherwise the error will be caught by the outer try catch
return await parser.parseString(content)
} catch (error) {
console.log(error)
return null
}
}
const sendUpdateSubscriptionMutation = async (
userId: string,
subscriptionId: string,
lastFetchedAt: Date,
lastFetchedChecksum: string,
scheduledAt: Date
) => {
const JWT_SECRET = env.server.jwtSecret
const REST_BACKEND_ENDPOINT = process.env.INTERNAL_API_URL
if (!JWT_SECRET || !REST_BACKEND_ENDPOINT) {
throw 'Environment not configured correctly'
}
const data = JSON.stringify({
query: `mutation UpdateSubscription($input: UpdateSubscriptionInput!){
updateSubscription(input:$input){
... on UpdateSubscriptionSuccess{
subscription{
id
lastFetchedAt
}
}
... on UpdateSubscriptionError{
errorCodes
}
}
}`,
variables: {
input: {
id: subscriptionId,
lastFetchedAt,
lastFetchedChecksum,
scheduledAt,
},
},
})
const auth = (await signToken({ uid: userId }, JWT_SECRET)) as string
try {
const response = await axios.post(
`${REST_BACKEND_ENDPOINT}/graphql`,
data,
{
headers: {
Cookie: `auth=${auth};`,
'Content-Type': 'application/json',
},
timeout: 30000, // 30s
}
)
/* eslint-disable @typescript-eslint/no-unsafe-member-access */
return !!response.data.data.updateSubscription.subscription
} catch (error) {
if (axios.isAxiosError(error)) {
console.error('update subscription mutation error', error.message)
} else {
console.error(error)
}
return false
}
}
const isItemRecentlySaved = async (userId: string, url: string) => {
const key = `recent-saved-item:${userId}:${url}`
try {
const result = await redisDataSource.redisClient?.get(key)
return !!result
} catch (err) {
console.error('error checking if item is old', err)
}
// If we failed to check, assume the item is good
return false
}
const createTask = async (
userId: string,
feedUrl: string,
item: RssFeedItem,
fetchContent: boolean,
folder: FolderType
) => {
const isRecentlySaved = await isItemRecentlySaved(userId, item.link)
if (isRecentlySaved) {
console.log('Item recently saved', item.link)
return true
}
if (folder === 'following' && !fetchContent) {
return createItemWithPreviewContent(userId, feedUrl, item)
}
return fetchContentAndCreateItem(userId, feedUrl, item, folder)
}
const fetchContentAndCreateItem = async (
userId: string,
feedUrl: string,
item: RssFeedItem,
folder: string
) => {
const payload = {
userId,
source: 'rss-feeder',
url: item.link.trim(),
saveRequestId: '',
labels: [{ name: 'RSS' }],
rssFeedUrl: feedUrl,
savedAt: item.isoDate,
publishedAt: item.isoDate,
folder,
}
try {
const task = await createHttpTaskWithToken({
queue: 'omnivore-rss-feed-queue',
taskHandlerUrl: env.queue.contentFetchGCFUrl,
payload,
})
return !!task
} catch (error) {
console.error('Error while creating task', error)
return false
}
}
const createItemWithPreviewContent = async (
userId: string,
feedUrl: string,
item: RssFeedItem
) => {
const input = {
userIds: [userId],
url: item.link,
title: item.title,
author: item.creator,
description: item.summary,
addedToFollowingFrom: 'feed',
previewContent: item.content || item.contentSnippet || item.summary,
addedToFollowingBy: feedUrl,
savedAt: item.isoDate,
publishedAt: item.isoDate,
previewContentType: 'text/html', // TODO: get content type from feed
thumbnail: getThumbnail(item),
}
try {
const serviceBaseUrl = process.env.INTERNAL_API_URL
const token = process.env.PUBSUB_VERIFICATION_TOKEN
if (!serviceBaseUrl || !token) {
throw 'Environment not configured correctly'
}
// save page
const taskHandlerUrl = `${serviceBaseUrl}svc/following/save?token=${token}`
const task = await createHttpTaskWithToken({
queue: env.queue.name,
priority: 'low',
taskHandlerUrl: taskHandlerUrl,
payload: input,
})
return !!task
} catch (error) {
console.error('Error while creating task', error)
return false
}
}
dotenv.config()
// Sentry.GCPFunction.init({
// dsn: process.env.SENTRY_DSN,
// tracesSampleRate: 0,
// })
const signToken = promisify(jwt.sign)
const parser = new Parser({
customFields: {
item: [
['link', 'links', { keepArray: true }],
'published',
'updated',
'created',
['media:content', 'media:content', { keepArray: true }],
['media:thumbnail'],
],
feed: [
'lastBuildDate',
'syn:updatePeriod',
'syn:updateFrequency',
'sy:updatePeriod',
'sy:updateFrequency',
],
},
})
const getUpdateFrequency = (feed: RssFeed) => {
const updateFrequency =
feed['syn:updateFrequency'] || feed['sy:updateFrequency']
if (!updateFrequency) {
return 1
}
const frequency = parseInt(updateFrequency, 10)
if (isNaN(frequency)) {
return 1
}
return frequency
}
const getUpdatePeriodInHours = (feed: RssFeed) => {
const updatePeriod = feed['syn:updatePeriod'] || feed['sy:updatePeriod']
switch (updatePeriod) {
case 'hourly':
return 1
case 'daily':
return 24
case 'weekly':
return 7 * 24
case 'monthly':
return 30 * 24
case 'yearly':
return 365 * 24
default:
// default to hourly
return 1
}
}
// get link following the order of preference: via, alternate, self
const getLink = (links: RssFeedItemLink[]): string | undefined => {
// sort links by preference
const sortedLinks: string[] = []
links.forEach((link) => {
// if link is a string, it is the href
if (typeof link === 'string') {
return sortedLinks.push(link)
}
if (link.$.rel === 'via') {
sortedLinks[0] = link.$.href
}
if (link.$.rel === 'alternate') {
sortedLinks[1] = link.$.href
}
if (link.$.rel === 'self' || !link.$.rel) {
sortedLinks[2] = link.$.href
}
})
// return the first link that is not undefined
return sortedLinks.find((link) => !!link)
}
const processSubscription = async (
subscriptionId: string,
userId: string,
feedUrl: string,
fetchResult: { content: string; checksum: string },
lastFetchedAt: number,
scheduledAt: number,
lastFetchedChecksum: string,
fetchContent: boolean,
folder: FolderType,
feed: RssFeed
) => {
let lastItemFetchedAt: Date | null = null
let lastValidItem: RssFeedItem | null = null
if (fetchResult.checksum === lastFetchedChecksum) {
console.log('feed has not been updated', feedUrl, lastFetchedChecksum)
return
}
const updatedLastFetchedChecksum = fetchResult.checksum
// fetch feed
let itemCount = 0
const feedLastBuildDate = feed.lastBuildDate
console.log('Feed last build date', feedLastBuildDate)
if (
feedLastBuildDate &&
new Date(feedLastBuildDate) <= new Date(lastFetchedAt)
) {
console.log('Skipping old feed', feedLastBuildDate)
return
}
// save each item in the feed
for (const item of feed.items) {
// use published or updated if isoDate is not available for atom feeds
const isoDate =
item.isoDate || item.published || item.updated || item.created
console.log('Processing feed item', item.links, item.isoDate, feed.feedUrl)
if (!item.links || item.links.length === 0) {
console.log('Invalid feed item', item)
continue
}
const link = getLink(item.links)
if (!link) {
console.log('Invalid feed item links', item.links)
continue
}
console.log('Fetching feed item', link)
const feedItem = {
...item,
isoDate,
link,
}
const publishedAt = feedItem.isoDate
? new Date(feedItem.isoDate)
: new Date()
// remember the last valid item
if (
!lastValidItem ||
(lastValidItem.isoDate && publishedAt > new Date(lastValidItem.isoDate))
) {
lastValidItem = feedItem
}
// Max limit per-feed update
if (itemCount > 99) {
continue
}
// skip old items
if (isOldItem(feedItem, lastFetchedAt)) {
console.log('Skipping old feed item', feedItem.link)
continue
}
const created = await createTask(
userId,
feedUrl,
feedItem,
fetchContent,
folder
)
if (!created) {
console.error('Failed to create task for feed item', feedItem.link)
continue
}
// remember the last item fetched at
if (!lastItemFetchedAt || publishedAt > lastItemFetchedAt) {
lastItemFetchedAt = publishedAt
}
itemCount = itemCount + 1
}
// no items saved
if (!lastItemFetchedAt) {
// the feed has been fetched before, no new valid items found
if (lastFetchedAt || !lastValidItem) {
console.log('No new valid items found')
return
}
// the feed has never been fetched, save at least the last valid item
const created = await createTask(
userId,
feedUrl,
lastValidItem,
fetchContent,
folder
)
if (!created) {
console.error('Failed to create task for feed item', lastValidItem.link)
throw new Error('Failed to create task for feed item')
}
lastItemFetchedAt = lastValidItem.isoDate
? new Date(lastValidItem.isoDate)
: new Date()
}
const updateFrequency = getUpdateFrequency(feed)
const updatePeriodInMs = getUpdatePeriodInHours(feed) * 60 * 60 * 1000
const nextScheduledAt = scheduledAt + updatePeriodInMs * updateFrequency
// update subscription lastFetchedAt
const updatedSubscription = await sendUpdateSubscriptionMutation(
userId,
subscriptionId,
lastItemFetchedAt,
updatedLastFetchedChecksum,
new Date(nextScheduledAt)
)
console.log('Updated subscription', updatedSubscription)
}
export const refreshFeed = async (request: any) => {
if (isRefreshFeedRequest(request)) {
return _refreshFeed(request)
}
console.log('not a feed to refresh')
return false
}
export const _refreshFeed = async (request: RefreshFeedRequest) => {
try {
const {
feedUrl,
subscriptionIds,
lastFetchedTimestamps,
scheduledTimestamps,
userIds,
lastFetchedChecksums,
fetchContents,
folders,
} = request
console.log('Processing feed', feedUrl)
const isBlocked = await isFeedBlocked(feedUrl)
if (isBlocked) {
console.log('feed is blocked: ', feedUrl)
return
}
const fetchResult = await fetchAndChecksum(feedUrl)
if (!fetchResult) {
console.error('Failed to fetch RSS feed', feedUrl)
await incrementFeedFailure(feedUrl)
return
}
const feed = await parseFeed(feedUrl, fetchResult.content)
if (!feed) {
console.error('Failed to parse RSS feed', feedUrl)
await incrementFeedFailure(feedUrl)
return
}
let allowFetchContent = true
if (isContentFetchBlocked(feedUrl)) {
console.log('fetching content blocked for feed: ', feedUrl)
allowFetchContent = false
}
console.log('Fetched feed', feed.title, new Date())
// process each subscription sequentially
for (let i = 0; i < subscriptionIds.length; i++) {
await processSubscription(
subscriptionIds[i],
userIds[i],
feedUrl,
fetchResult,
lastFetchedTimestamps[i],
scheduledTimestamps[i],
lastFetchedChecksums[i],
fetchContents[i] && allowFetchContent,
folders[i],
feed
)
}
} catch (e) {
console.error('Error while saving RSS feeds', e)
}
}

120
packages/api/src/queue-processor.ts Normal file
View file

@ -0,0 +1,120 @@
/* eslint-disable @typescript-eslint/no-floating-promises */
/* eslint-disable @typescript-eslint/restrict-template-expressions */
/* eslint-disable @typescript-eslint/require-await */
/* eslint-disable @typescript-eslint/no-misused-promises */
import express, { Express } from 'express'
import { appDataSource } from './data_source'
import { getEnv } from './util'
import { redisDataSource } from './redis_data_source'
import { CustomTypeOrmLogger } from './utils/logger'
import { SnakeNamingStrategy } from 'typeorm-naming-strategies'
import { refreshAllFeeds } from './jobs/rss/refreshAllFeeds'
import { Job, Worker, QueueEvents } from 'bullmq'
import { refreshFeed } from './jobs/rss/refreshFeed'
import { env } from './env'
export const QUEUE_NAME = 'omnivore-backend-queue'
const main = async () => {
console.log('[queue-processor]: starting queue processor')
const app: Express = express()
const port = process.env.PORT || 3002
redisDataSource.setOptions({
REDIS_URL: env.redis.url,
REDIS_CERT: env.redis.cert,
})
appDataSource.setOptions({
type: 'postgres',
host: env.pg.host,
port: env.pg.port,
schema: 'omnivore',
username: env.pg.userName,
password: env.pg.password,
database: env.pg.dbName,
logging: ['query', 'info'],
entities: [__dirname + '/entity/**/*{.js,.ts}'],
subscribers: [__dirname + '/events/**/*{.js,.ts}'],
namingStrategy: new SnakeNamingStrategy(),
logger: new CustomTypeOrmLogger(['query', 'info']),
connectTimeoutMS: 40000, // 40 seconds
maxQueryExecutionTime: 10000, // 10 seconds
})
// respond healthy to auto-scaler.
app.get('/_ah/health', (req, res) => res.sendStatus(200))
const server = app.listen(port, () => {
console.log(`[queue-processor]: started`)
})
// This is done after all the setup so it can access the
// environment that was loaded from GCP
await appDataSource.initialize()
await redisDataSource.initialize()
const redisClient = redisDataSource.redisClient
const workerRedisClient = redisDataSource.workerRedisClient
if (!workerRedisClient || !redisClient) {
throw '[queue-processor] error redis is not initialized'
}
const worker = new Worker(
QUEUE_NAME,
async (job: Job) => {
switch (job.name) {
case 'refresh-all-feeds': {
return await refreshAllFeeds(appDataSource)
}
case 'refresh-feed': {
return await refreshFeed(job.data)
}
}
return true
},
{
connection: workerRedisClient,
}
)
const queueEvents = new QueueEvents(QUEUE_NAME, {
connection: workerRedisClient,
})
queueEvents.on('added', async (job) => {
console.log('added job: ', job.jobId)
})
queueEvents.on('removed', async (job) => {
console.log('removed job: ', job.jobId)
})
queueEvents.on('completed', async (job) => {
console.log('completed job: ', job.jobId)
})
workerRedisClient.on('error', (error) => {
console.trace('[queue-processor]: redis worker error', { error })
})
redisClient.on('error', (error) => {
console.trace('[queue-processor]: redis error', { error })
})
const gracefulShutdown = async (signal: string) => {
console.log(`[queue-processor]: Received ${signal}, closing server...`)
await worker.close()
await redisDataSource.shutdown()
process.exit(0)
}
process.on('SIGINT', () => gracefulShutdown('SIGINT'))
process.on('SIGTERM', () => gracefulShutdown('SIGTERM'))
}
// only call main if the file was called from the CLI and wasn't required from another module
if (require.main === module) {
main().catch((e) => console.error(e))
}

39
packages/api/src/redis.ts
View file

@ -1,39 +0,0 @@
import { Redis } from 'ioredis'
import { env } from './env'
const url = env.redis.url
const cert = env.redis.cert
export const redisClient = url
? new Redis(url, {
connectTimeout: 10000, // 10 seconds
tls: cert
? {
cert,
rejectUnauthorized: false, // for self-signed certs
}
: undefined,
reconnectOnError: (err) => {
const targetErrors = [/READONLY/, /ETIMEDOUT/]
targetErrors.forEach((targetError) => {
if (targetError.test(err.message)) {
// Only reconnect when the error contains the keyword
return true
}
})
return false
},
retryStrategy: (times) => {
if (times > 10) {
// End reconnecting after a specific number of tries and flush all commands with a individual error
return null
}
// reconnect after
return Math.min(times * 50, 2000)
},
maxRetriesPerRequest: 1,
})
: null

98
packages/api/src/redis_data_source.ts Normal file
View file

@ -0,0 +1,98 @@
import Redis, { RedisOptions } from 'ioredis'
import { env } from './env'
export type RedisDataSourceOptions = {
REDIS_URL?: string
REDIS_CERT?: string
}
export class RedisDataSource {
options: RedisDataSourceOptions
isInitialized: boolean
redisClient: Redis | undefined = undefined
workerRedisClient: Redis | undefined = undefined
constructor(options: RedisDataSourceOptions) {
this.options = options
this.isInitialized = false
}
// Forcing this to be async as we might do some more initialization in the future
async initialize(): Promise<this> {
if (this.isInitialized) throw 'Error already initialized'
this.redisClient = createIORedisClient('app', this.options)
this.workerRedisClient = createIORedisClient('worker', this.options)
this.isInitialized = true
return Promise.resolve(this)
}
setOptions(options: RedisDataSourceOptions): void {
this.options = options
}
async shutdown(): Promise<void> {
this.isInitialized = false
try {
await this.workerRedisClient?.quit()
await this.redisClient?.quit()
} catch (err) {
console.error('error while shutting down redis')
}
}
}
const createIORedisClient = (
name: string,
options: RedisDataSourceOptions
): Redis | undefined => {
const redisURL = options.REDIS_URL
if (!redisURL) {
throw 'Error: no redisURL supplied'
}
const tls =
redisURL.startsWith('rediss://') && options.REDIS_CERT
? {
ca: options.REDIS_CERT,
rejectUnauthorized: false,
}
: undefined
const redisOptions: RedisOptions = {
tls,
name,
connectTimeout: 10000,
maxRetriesPerRequest: null,
offlineQueue: false,
// reconnectOnError: (err: Error) => {
// const targetErrors = [/READONLY/, /ETIMEDOUT/]
// targetErrors.forEach((targetError) => {
// if (targetError.test(err.message)) {
// // Only reconnect when the error contains the keyword
// return true
// }
// })
// return false
// },
retryStrategy: (times: number) => {
// if (times > 10) {
// // End reconnecting after a specific number of tries and flush all commands with a individual error
// return null
// }
// // reconnect after
// return Math.min(times * 50, 2000)
return 10
},
}
return new Redis(redisURL, redisOptions)
}
export const redisDataSource = new RedisDataSource({
REDIS_URL: env.redis.url,
REDIS_CERT: env.redis.cert,
})

56
packages/api/src/routers/svc/rss_feed.ts
View file

@ -1,16 +1,8 @@
/* eslint-disable @typescript-eslint/no-misused-promises */
import express from 'express'
import {
DEFAULT_SUBSCRIPTION_FOLDER,
Subscription,
} from '../../entity/subscription'
import { SubscriptionStatus, SubscriptionType } from '../../generated/graphql'
import { queueRSSRefreshAllFeedsJob } from '../../jobs/rss/refreshAllFeeds'
import { readPushSubscription } from '../../pubsub'
import { getRepository } from '../../repository'
import {
enqueueRssFeedFetch,
RssSubscriptionGroup,
} from '../../utils/createTask'
import { redisDataSource } from '../../redis_data_source'
import { logger } from '../../utils/logger'
export function rssFeedRouter() {
@ -28,44 +20,12 @@ export function rssFeedRouter() {
return res.status(200).send('Expired')
}
// get active rss feed subscriptions scheduled for fetch and group by feed url
const subscriptionGroups = (await getRepository(Subscription).query(
`
SELECT
url,
ARRAY_AGG(id) AS "subscriptionIds",
ARRAY_AGG(user_id) AS "userIds",
ARRAY_AGG(last_fetched_at) AS "fetchedDates",
ARRAY_AGG(coalesce(scheduled_at, NOW())) AS "scheduledDates",
ARRAY_AGG(last_fetched_checksum) AS checksums,
ARRAY_AGG(fetch_content) AS "fetchContents",
ARRAY_AGG(coalesce(folder, $3)) AS folders
FROM
omnivore.subscriptions
WHERE
type = $1
AND status = $2
AND (scheduled_at <= NOW() OR scheduled_at IS NULL)
GROUP BY
url
`,
[
SubscriptionType.Rss,
SubscriptionStatus.Active,
DEFAULT_SUBSCRIPTION_FOLDER,
]
)) as RssSubscriptionGroup[]
// create a cloud taks to fetch rss feed item for each subscription
await Promise.all(
subscriptionGroups.map((subscriptionGroup) => {
try {
return enqueueRssFeedFetch(subscriptionGroup)
} catch (error) {
logger.info('error creating rss feed fetch task', error)
}
})
)
if (redisDataSource.workerRedisClient) {
await queueRSSRefreshAllFeedsJob()
} else {
console.log('unable to fetchAll feeds, redis is not configured')
return res.status(500).send('Expired')
}
} catch (error) {
logger.info('error fetching rss feeds', error)
return res.status(500).send('Internal Server Error')

16
packages/api/src/server.ts
View file

@ -15,7 +15,6 @@ import { config, loggers } from 'winston'
import { makeApolloServer } from './apollo'
import { appDataSource } from './data_source'
import { env } from './env'
import { redisClient } from './redis'
import { articleRouter } from './routers/article_router'
import { authRouter } from './routers/auth/auth_router'
import { mobileAuthRouter } from './routers/auth/mobile/mobile_auth_router'
@ -45,6 +44,7 @@ import {
} from './utils/auth'
import { corsConfig } from './utils/corsConfig'
import { buildLogger, buildLoggerTransport } from './utils/logger'
import { redisDataSource } from './redis_data_source'
const PORT = process.env.PORT || 4000
@ -158,9 +158,9 @@ const main = async (): Promise<void> => {
// as healthy.
await appDataSource.initialize()
// redis is optional
if (redisClient) {
console.log('Redis Client Connected:', env.redis.url)
// redis is optional for the API server
if (env.redis.url) {
await redisDataSource.initialize()
}
const { app, apollo, httpServer } = createApp()
@ -190,10 +190,10 @@ const main = async (): Promise<void> => {
listener.timeout = 640 * 1000 // match headersTimeout
process.on('SIGINT', async () => {
if (redisClient) {
await redisClient.quit()
console.log('Redis client closed.')
}
// Shutdown redis before DB because the quit sequence can
// cause appDataSource to get reloaded in the callback
await redisDataSource.shutdown()
console.log('Redis connection closed.')
await appDataSource.destroy()
console.log('DB connection closed.')

6
packages/api/src/services/library_item.ts
View file

@ -8,7 +8,6 @@ import { Label } from '../entity/label'
import { LibraryItem, LibraryItemState } from '../entity/library_item'
import { BulkActionType, InputMaybe, SortParams } from '../generated/graphql'
import { createPubSubClient, EntityType } from '../pubsub'
import { redisClient } from '../redis'
import {
authTrx,
getColumns,
@ -18,6 +17,7 @@ import {
import { libraryItemRepository } from '../repository/library_item'
import { setRecentlySavedItemInRedis, wordsCount } from '../utils/helpers'
import { parseSearchQuery } from '../utils/search'
import { redisDataSource } from '../redis_data_source'
enum ReadFilter {
ALL = 'all',
@ -824,9 +824,9 @@ export const createLibraryItem = async (
)
// set recently saved item in redis if redis is enabled
if (redisClient) {
if (redisDataSource.redisClient) {
await setRecentlySavedItemInRedis(
redisClient,
redisDataSource.redisClient,
userId,
newLibraryItem.originalUrl
)

1
packages/api/src/services/save_page.ts
View file

@ -13,7 +13,6 @@ import {
SavePageInput,
SaveResult,
} from '../generated/graphql'
import { redisClient } from '../redis'
import { authTrx } from '../repository'
import { enqueueThumbnailTask } from '../utils/createTask'
import {

48
packages/api/src/util.ts
View file

@ -4,7 +4,7 @@
import * as dotenv from 'dotenv'
import os from 'os'
interface BackendEnv {
export interface BackendEnv {
pg: {
host: string
port: number
@ -25,8 +25,6 @@ interface BackendEnv {
}
client: {
url: string
previewGenerationServiceUrl: string
previewImageWrapperId: string
}
google: {
auth: {
@ -93,10 +91,6 @@ interface BackendEnv {
readwise: {
apiUrl: string
}
azure: {
speechKey: string
speechRegion: string
}
gcp: {
location: string
}
@ -115,19 +109,6 @@ interface BackendEnv {
}
}
/***
* Checks if we are running on Google App Engine.
* See https://cloud.google.com/appengine/docs/standard/nodejs/runtime#environment_variables
*/
export function isAppEngine(): boolean {
return (
process.env.GOOGLE_CLOUD_PROJECT !== undefined &&
process.env.GAE_INSTANCE !== undefined &&
process.env.GAE_SERVICE !== undefined &&
process.env.GAE_VERSION !== undefined
)
}
const nullableEnvVars = [
'INTERCOM_TOKEN',
'INTERCOM_SECRET_KEY',
@ -144,8 +125,6 @@ const nullableEnvVars = [
'PUPPETEER_QUEUE_NAME',
'CONTENT_FETCH_URL',
'CONTENT_FETCH_GCF_URL',
'PREVIEW_IMAGE_WRAPPER_ID',
'PREVIEW_GENERATION_SERVICE_URL',
'GCS_UPLOAD_SA_KEY_FILE_PATH',
'GAUTH_IOS_CLIENT_ID',
'GAUTH_ANDROID_CLIENT_ID',
@ -164,8 +143,6 @@ const nullableEnvVars = [
'READWISE_API_URL',
'INTEGRATION_TASK_HANDLER_URL',
'TEXT_TO_SPEECH_TASK_HANDLER_URL',
'AZURE_SPEECH_KEY',
'AZURE_SPEECH_REGION',
'GCP_LOCATION',
'RECOMMENDATION_TASK_HANDLER_URL',
'POCKET_CONSUMER_KEY',
@ -182,13 +159,8 @@ const nullableEnvVars = [
] // Allow some vars to be null/empty
/* If not in GAE and Prod/QA/Demo env (f.e. on localhost/dev env), allow following env vars to be null */
if (
!isAppEngine() &&
['prod', 'qa', 'demo'].indexOf(process.env.API_ENV || '') === -1
) {
nullableEnvVars.push(
...['GCS_UPLOAD_BUCKET', 'PREVIEW_GENERATION_SERVICE_URL']
)
if (process.env.API_ENV == 'local') {
nullableEnvVars.push(...['GCS_UPLOAD_BUCKET'])
}
const envParser =
@ -205,6 +177,10 @@ const envParser =
)
}
interface Dict<T> {
[key: string]: T | undefined
}
export function getEnv(): BackendEnv {
// Dotenv parses env file merging into proces.env which is then read into custom struct here.
dotenv.config()
@ -231,8 +207,6 @@ export function getEnv(): BackendEnv {
}
const client = {
url: parse('CLIENT_URL'),
previewGenerationServiceUrl: parse('PREVIEW_GENERATION_SERVICE_URL'),
previewImageWrapperId: parse('PREVIEW_IMAGE_WRAPPER_ID'),
}
const google = {
auth: {
@ -256,7 +230,7 @@ export function getEnv(): BackendEnv {
host: parse('JAEGER_HOST'),
}
const dev = {
isLocal: !isAppEngine(),
isLocal: parse('API_ENV') == 'local',
}
const queue = {
location: parse('PUPPETEER_QUEUE_LOCATION'),
@ -302,11 +276,6 @@ export function getEnv(): BackendEnv {
apiUrl: parse('READWISE_API_URL'),
}
const azure = {
speechKey: parse('AZURE_SPEECH_KEY'),
speechRegion: parse('AZURE_SPEECH_REGION'),
}
const gcp = {
location: parse('GCP_LOCATION'),
}
@ -344,7 +313,6 @@ export function getEnv(): BackendEnv {
sender,
sendgrid,
readwise,
azure,
gcp,
pocket,
subscription,

82
packages/api/src/utils/createTask.ts
View file

@ -19,6 +19,9 @@ import { generateVerificationToken, OmnivoreAuthorizationHeader } from './auth'
import { CreateTaskError } from './errors'
import { logger } from './logger'
import View = google.cloud.tasks.v2.Task.View
import { stringToHash } from './helpers'
import { queueRSSRefreshFeedJob } from '../jobs/rss/refreshAllFeeds'
import { redisDataSource } from '../redis_data_source'
// Instantiates a client.
const client = new CloudTasksClient()
@ -61,6 +64,9 @@ const createHttpTaskWithToken = async ({
> => {
// If there is no Google Cloud Project Id exposed, it means that we are in local environment
if (env.dev.isLocal || !project) {
console.error(
'error: attempting to create a cloud task but not running in google cloud.'
)
return null
}
@ -649,39 +655,55 @@ export const enqueueRssFeedFetch = async (
folders: subscriptionGroup.folders,
}
// If there is no Google Cloud Project Id exposed, it means that we are in local environment
if (env.dev.isLocal || !GOOGLE_CLOUD_PROJECT) {
if (env.queue.rssFeedTaskHandlerUrl) {
// Calling the handler function directly.
setTimeout(() => {
axios
.post(
`${env.queue.rssFeedTaskHandlerUrl}?token=${PUBSUB_VERIFICATION_TOKEN}`,
payload
)
.catch((error) => {
logError(error)
})
}, 0)
}
return nanoid()
}
let jobid = `refresh-feed_${stringToHash(
subscriptionGroup.url
)}_${stringToHash(JSON.stringify(subscriptionGroup.userIds.sort()))}`
const createdTasks = await createHttpTaskWithToken({
project: GOOGLE_CLOUD_PROJECT,
queue: 'omnivore-rss-queue',
payload,
taskHandlerUrl: `${env.queue.rssFeedTaskHandlerUrl}?token=${PUBSUB_VERIFICATION_TOKEN}`,
})
if (!createdTasks || !createdTasks[0].name) {
logger.error(`Unable to get the name of the task`, {
payload,
createdTasks,
if (redisDataSource.workerRedisClient) {
let job = await queueRSSRefreshFeedJob(jobid, payload, {
priority: 'high',
})
throw new CreateTaskError(`Unable to get the name of the task`)
if (!job || !job.id) {
throw 'unable to queue rss-refresh-feed-job, job did not enqueue'
}
return job.id
} else {
throw 'unable to queue rss-refresh-feed-job, redis is not configured'
}
return createdTasks[0].name
// // If there is no Google Cloud Project Id exposed, it means that we are in local environment
// if (env.dev.isLocal || !GOOGLE_CLOUD_PROJECT) {
// if (env.queue.rssFeedTaskHandlerUrl) {
// // Calling the handler function directly.
// setTimeout(() => {
// axios
// .post(
// `${env.queue.rssFeedTaskHandlerUrl}?token=${PUBSUB_VERIFICATION_TOKEN}`,
// payload
// )
// .catch((error) => {
// logError(error)
// })
// }, 0)
// }
// return nanoid()
// }
// const createdTasks = await createHttpTaskWithToken({
// project: GOOGLE_CLOUD_PROJECT,
// queue: 'omnivore-rss-queue',
// payload,
// taskHandlerUrl: `${env.queue.rssFeedTaskHandlerUrl}?token=${PUBSUB_VERIFICATION_TOKEN}`,
// })
// if (!createdTasks || !createdTasks[0].name) {
// logger.error(`Unable to get the name of the task`, {
// payload,
// createdTasks,
// })
// throw new CreateTaskError(`Unable to get the name of the task`)
// }
//return createdTasks[0].name
}
export default createHttpTaskWithToken

9
packages/api/src/utils/helpers.ts
View file

@ -26,7 +26,7 @@ import {
SearchItem,
} from '../generated/graphql'
import { createPubSubClient } from '../pubsub'
import { redisClient } from '../redis'
import { redisDataSource } from '../redis_data_source'
import { Claims, WithDataSourcesContext } from '../resolvers/types'
import { validateUrl } from '../services/create_page_save_request'
import { updateLibraryItem } from '../services/library_item'
@ -416,6 +416,13 @@ export const setRecentlySavedItemInRedis = async (
url: string
) => {
// save the url in redis for 26 hours so rss-feeder won't try to re-save it
if (!redisClient) {
console.info(
'not setting recently saved item because redis is not configured'
)
return
}
// save the url in redis for 8 hours so rss-feeder won't try to re-save it
const redisKey = `recent-saved-item:${userId}:${url}`
const ttlInSeconds = 60 * 60 * 26
try {

6
packages/api/test/global-setup.ts
View file

@ -1,4 +1,5 @@
import { redisClient } from '../src/redis'
import { env } from '../src/env'
import { redisDataSource } from '../src/redis_data_source'
import { createTestConnection } from './db'
import { startApolloServer } from './util'
@ -6,7 +7,8 @@ export const mochaGlobalSetup = async () => {
await createTestConnection()
console.log('db connection created')
if (redisClient) {
if (env.redis.url) {
await redisDataSource.initialize()
console.log('redis connection created')
}

7
packages/api/test/global-teardown.ts
View file

@ -1,5 +1,6 @@
import { appDataSource } from '../src/data_source'
import { redisClient } from '../src/redis'
import { env } from '../src/env'
import { redisDataSource } from '../src/redis_data_source'
import { stopApolloServer } from './util'
export const mochaGlobalTeardown = async () => {
@ -9,8 +10,8 @@ export const mochaGlobalTeardown = async () => {
await appDataSource.destroy()
console.log('db connection closed')
if (redisClient) {
await redisClient.quit()
if (env.redis.url) {
await redisDataSource.shutdown()
console.log('redis connection closed')
}
}

10
packages/api/test/routers/rss_feed.test.ts
View file

@ -4,9 +4,9 @@ import sinon from 'sinon'
import sinonChai from 'sinon-chai'
import { User } from '../../src/entity/user'
import { SubscriptionType } from '../../src/generated/graphql'
import * as refreshAllFeeds from '../../src/jobs/rss/refreshAllFeeds'
import { createRssSubscriptions } from '../../src/services/subscriptions'
import { deleteUser } from '../../src/services/user'
import * as createTask from '../../src/utils/createTask'
import { createTestUser } from '../db'
import { request } from '../util'
@ -80,11 +80,11 @@ describe('Rss feeds Router', () => {
},
}
// fake enqueueRssFeedFetch function
// fake queueRSSRefreshAllFeedsJob function
const fake = sinon.replace(
createTask,
'enqueueRssFeedFetch',
sinon.fake.resolves('task name')
refreshAllFeeds,
'queueRSSRefreshAllFeedsJob',
sinon.fake()
)
const res = await request

4
packages/text-to-speech/src/index.ts
View file

@ -241,8 +241,8 @@ export const textToSpeechStreamingHandler = Sentry.GCPFunction.wrapHttpFunction(
// create redis client
const redisClient = createRedisClient(
process.env.REDIS_URL,
process.env.REDIS_CERT
process.env.REDIS_TTS_URL,
process.env.REDIS_TTS_CERT
)
try {

2
packages/web/styles/articleInnerStyling.css
View file

@ -335,7 +335,7 @@ on smaller screens we display the note icon
vertical-align: bottom;
word-wrap: initial;
font-family: 'SF Mono', monospace !important;
white-space: pre;
white-space: break-spaces;
direction: ltr;
unicode-bidi: embed;
color: var(--font-color);

1
pkg/admin/src/index.ts
View file

@ -56,7 +56,6 @@ const ADMIN_USER_EMAIL =
const router = AdminJsExpress.buildAuthenticatedRouter(adminBro, {
authenticate: async (email, password) => {
const user = await AdminUser.findOne({ email })
console.log('looked up user: ', user)
if (user) {
const matched = await compare(password, user.password)
console.log(' -- failed match')

17
pkg/bull-queue-admin/Dockerfile Normal file
View file

@ -0,0 +1,17 @@
FROM node:18.16
WORKDIR /app
COPY package.json .
COPY yarn.lock .
RUN yarn install --frozen-lockfile
COPY . .
ENV PORT=8080
EXPOSE 8080
ENV NODE_ENV production
CMD yarn start

114
pkg/bull-queue-admin/index.js Normal file
View file

@ -0,0 +1,114 @@
const { createBullBoard } = require('@bull-board/api')
const { BullMQAdapter } = require('@bull-board/api/bullMQAdapter')
const { ExpressAdapter } = require('@bull-board/express')
const { Queue } = require('bullmq')
const { Redis } = require('ioredis')
const session = require('express-session')
const passport = require('passport')
const LocalStrategy = require('passport-local').Strategy
const { ensureLoggedIn } = require('connect-ensure-login')
const express = require('express')
const bodyParser = require('body-parser')
const readYamlFile = require('read-yaml-file')
passport.use(
new LocalStrategy(function (username, password, cb) {
readYamlFile(process.env.SECRETS_FILE).then((secrets) => {
if (
secrets.ADMIN_USER_PASSWORD &&
username === secrets.ADMIN_USER_EMAIL &&
password === secrets.ADMIN_USER_PASSWORD
) {
return cb(null, { user: 'bull-board' })
}
return cb(null, false)
})
})
)
// Configure Passport authenticated session persistence.
//
// In order to restore authentication state across HTTP requests, Passport needs
// to serialize users into and deserialize users out of the session. The
// typical implementation of this is as simple as supplying the user ID when
// serializing, and querying the user record by ID from the database when
// deserializing.
passport.serializeUser((user, cb) => {
cb(null, user)
})
passport.deserializeUser((user, cb) => {
cb(null, user)
})
const run = async () => {
const secrets = await readYamlFile(process.env.SECRETS_FILE)
const redisOptions = (secrets) => {
if (secrets.REDIS_URL?.startsWith('rediss://') && secrets.REDIS_CERT) {
return {
tls: {
ca: secrets.REDIS_CERT,
rejectUnauthorized: false,
},
}
}
return {}
}
const connection = new Redis(secrets.REDIS_URL, redisOptions(secrets))
console.log('set connection: ', connection)
const rssRefreshFeed = new Queue('omnivore-backend-queue', {
connection: connection,
})
const serverAdapter = new ExpressAdapter()
serverAdapter.setBasePath('/ui')
createBullBoard({
queues: [new BullMQAdapter(rssRefreshFeed)],
serverAdapter,
})
const app = express()
// Configure view engine to render EJS templates.
app.set('views', __dirname + '/views')
app.set('view engine', 'ejs')
app.use(
session({ secret: 'keyboard cat', saveUninitialized: true, resave: true })
)
app.use(bodyParser.urlencoded({ extended: false }))
// Initialize Passport and restore authentication state, if any, from the session.
app.use(passport.initialize({}))
app.use(passport.session({}))
app.get('/ui/login', (req, res) => {
res.render('login', { invalid: req.query.invalid === 'true' })
})
app.post(
'/ui/login',
passport.authenticate('local', {
failureRedirect: '/ui/login?invalid=true',
}),
(req, res) => {
res.redirect('/ui')
}
)
app.use(
'/ui',
ensureLoggedIn({ redirectTo: '/ui/login' }),
serverAdapter.getRouter()
)
app.listen(8080, () => {
console.log('Running on 8080...')
})
}
// eslint-disable-next-line no-console
run().catch((e) => console.error(e))

22
pkg/bull-queue-admin/package.json Normal file
View file

@ -0,0 +1,22 @@
{
"name": "queue-admin",
"version": "1.0.0",
"description": "Custom version of bull-board for adm",
"main": "index.js",
"scripts": {
"start": "node index.js",
"test": "echo \"Error: no test specified\" && exit 1"
},
"dependencies": {
"@bull-board/express": "^3.10.4",
"body-parser": "^1.20.2",
"bullmq": "^4.6.0",
"connect-ensure-login": "^0.1.1",
"express": "^4.17.3",
"express-session": "^1.17.2",
"ioredis": "^5.3.2",
"passport": "^0.6.0",
"passport-local": "^1.0.0",
"read-yaml-file": "^2.1.0"
}
}

86
pkg/bull-queue-admin/views/login.ejs Normal file
View file

@ -0,0 +1,86 @@
<link href="https://fonts.googleapis.com/css2?family=Ubuntu:wght@300;400;500&display=swap" rel="stylesheet"/>
<style>
body {
background: #f5f8fa;
font-family: 'Ubuntu', sans-serif;
font-weight: 400;
line-height: 1.25em;
margin: 0;
font-size: 16px;
color: #454b52;
-webkit-font-smoothing: antialiased;
-moz-osx-font-smoothing: grayscale;
}
.login-page {
width: 360px;
padding: 8% 0 0;
margin: auto;
}
.form {
position: relative;
z-index: 1;
background: #FFFFFF;
max-width: 360px;
margin: 0 auto 100px;
padding: 45px;
text-align: center;
box-shadow: 0 0 20px 0 rgba(0, 0, 0, 0.2), 0 5px 5px 0 rgba(0, 0, 0, 0.24);
}
.form input {
font-family: inherit;
outline: 0;
background: #f2f2f2;
width: 100%;
border: 0;
margin: 0 0 15px;
padding: 15px;
box-sizing: border-box;
font-size: 14px;
}
.form button {
font-family: inherit;
text-transform: uppercase;
outline: 0;
background: hsl(217, 22%, 24%);
width: 100%;
border: 0;
padding: 15px;
color: #FFFFFF;
font-size: 14px;
-webkit-transition: all 0.3 ease;
transition: all 0.3 ease;
cursor: pointer;
}
.form button:hover,.form button:active,.form button:focus {
background: hsl(217, 22%, 28%);
}
.form .message {
margin: 15px 0 0;
color: #b3b3b3;
font-size: 12px;
}
.form .message a {
color: hsl(217, 22%, 24%);
text-decoration: none;
}
.message.error {
color: #EF3B3A;
}
</style>
<div class="login-page">
<div class="form">
<form class="login-form" method="post" action="/ui/login">
<input type="text" name="username" placeholder="Username"/>
<input type="password" name="password" placeholder="Password"/>
<button>Login</button>
<p class="message">Username: bull, Password: board</p>
<% if (invalid) { %>
<p class="message error">Invalid username or password.</p>
<% } %>
</form>
</div>
</div>

1072
pkg/bull-queue-admin/yarn.lock Normal file
View file

File diff suppressed because it is too large Load diff

96
yarn.lock
View file

@ -2738,6 +2738,13 @@
lodash.snakecase "^4.1.1"
p-defer "^3.0.0"
"@google-cloud/secret-manager@^5.0.1":
version "5.0.1"
resolved "https://registry.yarnpkg.com/@google-cloud/secret-manager/-/secret-manager-5.0.1.tgz#8ea03952068c5982f5ba23776b268fade2bae425"
integrity sha512-elwopNbJhDBYdvCL8V0mbC+8UfXg/ElFosE1uRg7oKzRZJhUchsGE8Cxj1av807UkZvc4yzWCFg6/UXVXON3Kg==
dependencies:
google-gax "^4.0.3"
"@google-cloud/storage@^6.9.5":
version "6.12.0"
resolved "https://registry.yarnpkg.com/@google-cloud/storage/-/storage-6.12.0.tgz#a5d3093cc075252dca5bd19a3cfda406ad3a9de1"
@ -5980,6 +5987,11 @@
resolved "https://registry.yarnpkg.com/@sqltools/formatter/-/formatter-1.2.3.tgz#1185726610acc37317ddab11c3c7f9066966bd20"
integrity sha512-O3uyB/JbkAEMZaP3YqyHH7TMnex7tWyCbCI4EfJdOCoN6HIhqdJBWTM6aCCiWQ/5f5wxjgU735QAIpJbjDvmzg==
"@sqltools/formatter@^1.2.5":
version "1.2.5"
resolved "https://registry.yarnpkg.com/@sqltools/formatter/-/formatter-1.2.5.tgz#3abc203c79b8c3e90fd6c156a0c62d5403520e12"
integrity sha512-Uy0+khmZqUrUGm5dmMqVlnvufZRSK0FbYzVgp0UMstm+F5+W2/jnEEQyc9vo1ZR/E5ZI/B1WjjoTqBqwJL6Krw==
"@stitches/react@^1.2.5":
version "1.2.8"
resolved "https://registry.yarnpkg.com/@stitches/react/-/react-1.2.8.tgz#954f8008be8d9c65c4e58efa0937f32388ce3a38"
@ -7390,6 +7402,16 @@
"@types/qs" "*"
"@types/serve-static" "*"
"@types/express@^4.17.21":
version "4.17.21"
resolved "https://registry.yarnpkg.com/@types/express/-/express-4.17.21.tgz#c26d4a151e60efe0084b23dc3369ebc631ed192d"
integrity sha512-ejlPM315qwLpaQlQDTjPdsUFSc6ZsP4AN6AlWnogPjQ7CVi7PYF3YVz+CY3jE2pwYf7E/7HlDAN0rV2GxTG0HQ==
dependencies:
"@types/body-parser" "*"
"@types/express-serve-static-core" "^4.17.33"
"@types/qs" "*"
"@types/serve-static" "*"
"@types/filesystem@*":
version "0.0.32"
resolved "https://registry.yarnpkg.com/@types/filesystem/-/filesystem-0.0.32.tgz#307df7cc084a2293c3c1a31151b178063e0a8edf"
@ -7791,6 +7813,13 @@
resolved "https://registry.yarnpkg.com/@types/node/-/node-16.18.68.tgz#3155f64a961b3d8d10246c80657f9a7292e3421a"
integrity sha512-sG3hPIQwJLoewrN7cr0dwEy+yF5nD4D/4FxtQpFciRD/xwUzgD+G05uxZHv5mhfXo4F9Jkp13jjn0CC2q325sg==
"@types/node@^20.11.0":
version "20.11.0"
resolved "https://registry.yarnpkg.com/@types/node/-/node-20.11.0.tgz#8e0b99e70c0c1ade1a86c4a282f7b7ef87c9552f"
integrity sha512-o9bjXmDNcF7GbM4CNQpmi+TutCgap/K3w1JyKgxAjqx41zp9qlIAVFi0IhCNsJcXolEqLWhbFbEeL0PvYm4pcQ==
dependencies:
undici-types "~5.26.4"
"@types/nodemailer@^6.4.4":
version "6.4.4"
resolved "https://registry.yarnpkg.com/@types/nodemailer/-/nodemailer-6.4.4.tgz#c265f7e7a51df587597b3a49a023acaf0c741f4b"
@ -9315,6 +9344,11 @@ app-root-path@^3.0.0:
resolved "https://registry.yarnpkg.com/app-root-path/-/app-root-path-3.0.0.tgz#210b6f43873227e18a4b810a032283311555d5ad"
integrity sha512-qMcx+Gy2UZynHjOHOIXPNvpf+9cjvk3cWrBBK7zg4gH9+clobJRb9NGzcT7mQTcV/6Gm/1WelUtqxVXnNlrwcw==
app-root-path@^3.1.0:
version "3.1.0"
resolved "https://registry.yarnpkg.com/app-root-path/-/app-root-path-3.1.0.tgz#5971a2fc12ba170369a7a1ef018c71e6e47c2e86"
integrity sha512-biN3PwB2gUtjaYy/isrU3aNWI5w+fAfvHkSvCKeQGxhmYpwKFUxudR3Yya+KqVRHBmEDYh+/lTozYCFbmzX4nA==
apparatus@^0.0.10:
version "0.0.10"
resolved "https://registry.yarnpkg.com/apparatus/-/apparatus-0.0.10.tgz#81ea756772ada77863db54ceee8202c109bdca3e"
@ -12724,7 +12758,7 @@ dateformat@^3.0.0, dateformat@^3.0.3:
resolved "https://registry.yarnpkg.com/dateformat/-/dateformat-3.0.3.tgz#a6e37499a4d9a9cf85ef5872044d62901c9889ae"
integrity sha512-jyCETtSl3VMZMWeRo7iY1FL19ges1t55hMo5yaam4Jrsm5EPL89UQkoQRyiI+Yf4k8r2ZpdngkV8hr1lIdjb3Q==
dayjs@1.x, dayjs@^1.10.4, dayjs@^1.11.7:
dayjs@1.x, dayjs@^1.10.4, dayjs@^1.11.7, dayjs@^1.11.9:
version "1.11.10"
resolved "https://registry.yarnpkg.com/dayjs/-/dayjs-1.11.10.tgz#68acea85317a6e164457d6d6947564029a6a16a0"
integrity sha512-vjAczensTgRcqDERK0SR2XMwsF/tSvnvlv6VcF2GIhg6Sx4yOIt/irsr1RDJsKiIyBzJDpCoXiWWq28MqH2cnQ==
@ -13452,16 +13486,16 @@ dotenv@^16.0.1:
resolved "https://registry.yarnpkg.com/dotenv/-/dotenv-16.0.1.tgz#8f8f9d94876c35dac989876a5d3a82a267fdce1d"
integrity sha512-1K6hR6wtk2FviQ4kEiSjFiH5rpzEVi8WW0x96aztHVMhEspNpc4DVOUTEHtEva5VThQ8IaBX1Pe4gSzpVVUsKQ==
dotenv@^16.0.3, dotenv@~16.3.1:
version "16.3.1"
resolved "https://registry.yarnpkg.com/dotenv/-/dotenv-16.3.1.tgz#369034de7d7e5b120972693352a3bf112172cc3e"
integrity sha512-IPzF4w4/Rd94bA9imS68tZBaYyBWSCE47V1RGuMrB94iyTOIEwRmVL2x/4An+6mETpLrKJ5hQkB8W4kFAadeIQ==
dotenv@^8.0.0, dotenv@^8.2.0:
version "8.6.0"
resolved "https://registry.yarnpkg.com/dotenv/-/dotenv-8.6.0.tgz#061af664d19f7f4d8fc6e4ff9b584ce237adcb8b"
integrity sha512-IrPdXQsk2BbzvCBGBOTmmSH5SodmqZNt4ERAZDmW4CT+tL8VtvinqywuANaFu4bOMWki16nqf0e4oC0QIaDr/g==
dotenv@~16.3.1:
version "16.3.1"
resolved "https://registry.yarnpkg.com/dotenv/-/dotenv-16.3.1.tgz#369034de7d7e5b120972693352a3bf112172cc3e"
integrity sha512-IPzF4w4/Rd94bA9imS68tZBaYyBWSCE47V1RGuMrB94iyTOIEwRmVL2x/4An+6mETpLrKJ5hQkB8W4kFAadeIQ==
downshift@^6.0.15:
version "6.1.7"
resolved "https://registry.yarnpkg.com/downshift/-/downshift-6.1.7.tgz#fdb4c4e4f1d11587985cd76e21e8b4b3fa72e44c"
@ -14732,7 +14766,7 @@ express-rate-limit@^6.3.0:
resolved "https://registry.yarnpkg.com/express-rate-limit/-/express-rate-limit-6.11.1.tgz#52e05c5d379cd5d06ae29665862436eb712e414a"
integrity sha512-8+UpWtQY25lJaa4+3WxDBGDcAu4atcTruSs3QSL5VPEplYy6kmk84wutG9rUkkK5LmMQQ7TFHWLZYITwVNbbEg==
express@^4.16.4, express@^4.17.1:
express@^4.16.4, express@^4.17.1, express@^4.18.2:
version "4.18.2"
resolved "https://registry.yarnpkg.com/express/-/express-4.18.2.tgz#3fabe08296e930c796c19e3c516979386ba9fd59"
integrity sha512-5/PsL6iGPdfQ/lKM1UuielYgv3BUoJfz1aUwU9vHZ+J7gyvwdQXFEBIEIaxeGf0GIcreATNyBExtalisDbuMqQ==
@ -16083,7 +16117,7 @@ glob@7.2.0, glob@^7.1.1, glob@^7.1.2, glob@^7.1.3, glob@^7.1.4, glob@^7.1.6, glo
once "^1.3.0"
path-is-absolute "^1.0.0"
glob@^10.2.2:
glob@^10.2.2, glob@^10.3.10:
version "10.3.10"
resolved "https://registry.yarnpkg.com/glob/-/glob-10.3.10.tgz#0351ebb809fd187fe421ab96af83d3a70715df4b"
integrity sha512-fa46+tv1Ak0UPK1TOy/pZrIybNNt4HCv7SDzwyfiOZkvZLEbjsZkJBPtDHVshZjbecAoAGSC20MjLDG/qr679g==
@ -21777,6 +21811,11 @@ mkdirp@^1.0.3, mkdirp@^1.0.4:
resolved "https://registry.yarnpkg.com/mkdirp/-/mkdirp-1.0.4.tgz#3eb5ed62622756d79a5f0e2a221dfebad75c2f7e"
integrity sha512-vVqVZQyf3WLx2Shd0qJ9xuvqgAyKPLAiqITEtqW0oIUjzo3PePDd6fW9iFz30ef7Ysp/oiWqbhszeGWW2T6Gzw==
mkdirp@^2.1.3:
version "2.1.6"
resolved "https://registry.yarnpkg.com/mkdirp/-/mkdirp-2.1.6.tgz#964fbcb12b2d8c5d6fbc62a963ac95a273e2cc19"
integrity sha512-+hEnITedc8LAtIP9u3HJDFIdcLV2vXP33sqLLIzkv1Db1zO/1OxbvYf0Y1OC/S/Qo5dxHXepofhmxL02PsKe+A==
mkdirp@~0.3.5:
version "0.3.5"
resolved "https://registry.yarnpkg.com/mkdirp/-/mkdirp-0.3.5.tgz#de3e5f8961c88c787ee1368df849ac4413eca8d7"
@ -25964,6 +26003,14 @@ read-pkg@^7.1.0:
parse-json "^5.2.0"
type-fest "^2.0.0"
read-yaml-file@^2.1.0:
version "2.1.0"
resolved "https://registry.yarnpkg.com/read-yaml-file/-/read-yaml-file-2.1.0.tgz#c5866712db9ef5343b4d02c2413bada53c41c4a9"
integrity sha512-UkRNRIwnhG+y7hpqnycCL/xbTk7+ia9VuVTC0S+zVbwd65DI9eUpRMfsWIGrCWxTU/mi+JW8cHQCrv+zfCbEPQ==
dependencies:
js-yaml "^4.0.0"
strip-bom "^4.0.0"
read@1, read@^1.0.7, read@~1.0.7:
version "1.0.7"
resolved "https://registry.yarnpkg.com/read/-/read-1.0.7.tgz#b3da19bd052431a97671d44a42634adf710b40c4"
@ -29124,7 +29171,7 @@ tslib@^1.0.0, tslib@^1.8.1, tslib@^1.9.0, tslib@^1.9.3:
resolved "https://registry.yarnpkg.com/tslib/-/tslib-1.14.1.tgz#cf2d38bdc34a134bcaf1091c41f6619e2f672d00"
integrity sha512-Xni35NKzjgMrwevysHTCArtLDpPvye8zV/0E4EyYn43P7/7qvQwPh9BGkHewbMulVntbigmcT7rdX3BNo9wRJg==
tslib@^2, tslib@^2.0.0, tslib@^2.0.1, tslib@^2.0.3, tslib@^2.1.0, tslib@^2.3.0, tslib@^2.3.1, tslib@^2.4.0:
tslib@^2, tslib@^2.0.0, tslib@^2.0.1, tslib@^2.0.3, tslib@^2.1.0, tslib@^2.3.0, tslib@^2.3.1, tslib@^2.4.0, tslib@^2.5.0:
version "2.6.2"
resolved "https://registry.yarnpkg.com/tslib/-/tslib-2.6.2.tgz#703ac29425e7b37cd6fd456e92404d46d1f3e4ae"
integrity sha512-AEYxH93jGFPn/a2iVAwW87VuUIkR1FVUKB77NwMF7nBTDkDrrT/Hpt/IrCJ0QXhW27jTBDcf5ZY7w6RiqTMw2Q==
@ -29340,6 +29387,27 @@ typeorm-naming-strategies@^4.1.0:
resolved "https://registry.yarnpkg.com/typeorm-naming-strategies/-/typeorm-naming-strategies-4.1.0.tgz#1ec6eb296c8d7b69bb06764d5b9083ff80e814a9"
integrity sha512-vPekJXzZOTZrdDvTl1YoM+w+sUIfQHG4kZTpbFYoTsufyv9NIBRe4Q+PdzhEAFA2std3D9LZHEb1EjE9zhRpiQ==
typeorm@^0.3.19:
version "0.3.19"
resolved "https://registry.yarnpkg.com/typeorm/-/typeorm-0.3.19.tgz#a985ce8ae36d266018e44fed5e27a4a5da34ad2a"
integrity sha512-OGelrY5qEoAU80mR1iyvmUHiKCPUydL6xp6bebXzS7jyv/X70Gp/jBWRAfF4qGOfy2A7orMiGRfwsBUNbEL65g==
dependencies:
"@sqltools/formatter" "^1.2.5"
app-root-path "^3.1.0"
buffer "^6.0.3"
chalk "^4.1.2"
cli-highlight "^2.1.11"
dayjs "^1.11.9"
debug "^4.3.4"
dotenv "^16.0.3"
glob "^10.3.10"
mkdirp "^2.1.3"
reflect-metadata "^0.1.13"
sha.js "^2.4.11"
tslib "^2.5.0"
uuid "^9.0.0"
yargs "^17.6.2"
typeorm@^0.3.4:
version "0.3.7"
resolved "https://registry.yarnpkg.com/typeorm/-/typeorm-0.3.7.tgz#5776ed5058f0acb75d64723b39ff458d21de64c1"
@ -29378,6 +29446,11 @@ typescript@^4.4.4:
resolved "https://registry.yarnpkg.com/typescript/-/typescript-4.9.5.tgz#095979f9bcc0d09da324d58d03ce8f8374cbe65a"
integrity sha512-1FXk9E2Hm+QzZQ7z+McJiHL4NW1F2EzMu9Nq9i3zAaGqibafqYwCVU6WyWAuyQRRzOlxou8xZSyXLEN8oKj24g==
typescript@^5.3.3:
version "5.3.3"
resolved "https://registry.yarnpkg.com/typescript/-/typescript-5.3.3.tgz#b3ce6ba258e72e6305ba66f5c9b452aaee3ffe37"
integrity sha512-pXWcraxM0uxAS+tN0AG/BF2TyqmHO014Z070UsJ+pFvYuRSq8KH8DmWpnbXe0pEPDHXZV3FcAbJkijJ5oNEnWw==
ua-parser-js@^0.7.30:
version "0.7.33"
resolved "https://registry.yarnpkg.com/ua-parser-js/-/ua-parser-js-0.7.33.tgz#1d04acb4ccef9293df6f70f2c3d22f3030d8b532"
@ -29456,6 +29529,11 @@ underscore@^1.13.4, underscore@^1.9.1:
resolved "https://registry.yarnpkg.com/underscore/-/underscore-1.13.4.tgz#7886b46bbdf07f768e0052f1828e1dcab40c0dee"
integrity sha512-BQFnUDuAQ4Yf/cYY5LNrK9NCJFKriaRbD9uR1fTeXnBeoa97W0i41qkZfGO9pSo8I5KzjAcSY2XYtdf0oKd7KQ==
undici-types@~5.26.4:
version "5.26.5"
resolved "https://registry.yarnpkg.com/undici-types/-/undici-types-5.26.5.tgz#bcd539893d00b56e964fd2657a4866b221a65617"
integrity sha512-JlCMO+ehdEIKqlFxk6IfVoAUVmgz7cU7zD/h9XZ0qzeosSHmUJVOzSQvvYSYWXkFXC+IfLKSIffhv0sVZup6pA==
undici@^4.9.3:
version "4.14.1"
resolved "https://registry.yarnpkg.com/undici/-/undici-4.14.1.tgz#7633b143a8a10d6d63335e00511d071e8d52a1d9"