diff --git a/packages/api/.env.example b/packages/api/.env.example index 899b04dc3..f24c9c361 100644 --- a/packages/api/.env.example +++ b/packages/api/.env.example @@ -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 diff --git a/packages/api/.env.test b/packages/api/.env.test index 9b34455c1..52cc7075b 100644 --- a/packages/api/.env.test +++ b/packages/api/.env.test @@ -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/ diff --git a/packages/api/package.json b/packages/api/package.json index 432de1b5d..9060f0641 100644 --- a/packages/api/package.json +++ b/packages/api/package.json @@ -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" } -} +} \ No newline at end of file diff --git a/packages/api/src/jobs/rss/refreshAllFeeds.ts b/packages/api/src/jobs/rss/refreshAllFeeds.ts new file mode 100644 index 000000000..c9b93f196 --- /dev/null +++ b/packages/api/src/jobs/rss/refreshAllFeeds.ts @@ -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 => { + 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 => { + const queue = createBackendQueue() + if (!queue) { + return undefined + } + return queue.add('refresh-feed', payload, { + jobId: jobid, + removeOnComplete: true, + removeOnFail: true, + lifo: options.priority == 'high', + }) +} diff --git a/packages/api/src/jobs/rss/refreshFeed.ts b/packages/api/src/jobs/rss/refreshFeed.ts new file mode 100644 index 000000000..84cd82a75 --- /dev/null +++ b/packages/api/src/jobs/rss/refreshFeed.ts @@ -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