From 44473ba0898a9af4d9aa89ba68c798015b3cd2ae Mon Sep 17 00:00:00 2001 From: Hongbo Wu Date: Tue, 11 Jul 2023 12:22:49 +0800 Subject: [PATCH] upload feed subscriptions in cloud storage in cronjob --- packages/api/src/elastic/types.ts | 1 + packages/api/src/routers/svc/rss_feed.ts | 41 +++++++++++------ packages/rss-handler/package.json | 1 + packages/rss-handler/src/index.ts | 54 +++++++++++------------ packages/rss-handler/src/task.ts | 56 ++++++++++++++++++++++++ 5 files changed, 113 insertions(+), 40 deletions(-) create mode 100644 packages/rss-handler/src/task.ts diff --git a/packages/api/src/elastic/types.ts b/packages/api/src/elastic/types.ts index 320604c3b..b6e5bf41b 100644 --- a/packages/api/src/elastic/types.ts +++ b/packages/api/src/elastic/types.ts @@ -162,6 +162,7 @@ export interface Page { listenedAt?: Date wordsCount?: number recommendations?: Recommendation[] + rssFeedUrl?: string } export interface SearchItem { diff --git a/packages/api/src/routers/svc/rss_feed.ts b/packages/api/src/routers/svc/rss_feed.ts index a23f1acf0..e5a96af71 100644 --- a/packages/api/src/routers/svc/rss_feed.ts +++ b/packages/api/src/routers/svc/rss_feed.ts @@ -1,10 +1,12 @@ /* eslint-disable @typescript-eslint/no-misused-promises */ +import { stringify } from 'csv-stringify/.' import express from 'express' +import { DateTime } from 'luxon' import { readPushSubscription } from '../../datalayer/pubsub' import { Subscription } from '../../entity/subscription' import { getRepository } from '../../entity/utils' import { SubscriptionStatus, SubscriptionType } from '../../generated/graphql' -import { enqueueRssFeedFetch } from '../../utils/createTask' +import { createGCSFile } from '../../utils/uploads' export function rssFeedRouter() { const router = express.Router() @@ -20,9 +22,11 @@ export function rssFeedRouter() { return res.status(200).send('Expired') } + let writeStream: NodeJS.WritableStream | undefined try { // get all active rss feed subscriptions const subscriptions = await getRepository(Subscription).find({ + select: ['id', 'url', 'user'], where: { type: SubscriptionType.Rss, status: SubscriptionStatus.Active, @@ -30,22 +34,33 @@ export function rssFeedRouter() { relations: ['user'], }) - // create a cloud taks to fetch rss feed item for each subscription - await Promise.all( - subscriptions.map((subscription) => { - try { - return enqueueRssFeedFetch(subscription) - } catch (error) { - console.log('error creating rss feed fetch task', error) - } - }) - ) + // write the list of subscriptions to a csv file and upload it to gcs + // path style: rss/.csv + const dateStr = DateTime.now().toISODate() + const fullPath = `rss/${dateStr}.csv` + // open a write_stream to the file + const file = createGCSFile(fullPath) + writeStream = file.createWriteStream({ + contentType: 'text/csv', + }) + // stringify the data and pipe it to the write_stream + const stringifier = stringify({ + header: false, + columns: ['subscriptionId', 'userId', 'feedUrl'], + }) + stringifier.pipe(writeStream) - res.send('OK') + subscriptions.forEach((sub) => { + stringifier.write([sub.id, sub.user.id, sub.url]) + }) } catch (error) { console.log('error fetching rss feeds', error) - res.status(500).send('Internal Server Error') + return res.status(500).send('Internal Server Error') + } finally { + writeStream?.end() } + + res.send('OK') }) return router diff --git a/packages/rss-handler/package.json b/packages/rss-handler/package.json index e24080ae1..82b87db8c 100644 --- a/packages/rss-handler/package.json +++ b/packages/rss-handler/package.json @@ -21,6 +21,7 @@ }, "dependencies": { "@google-cloud/functions-framework": "3.1.2", + "@google-cloud/tasks": "^3.0.5", "@sentry/serverless": "^6.16.1", "axios": "^1.4.0", "dotenv": "^16.0.1", diff --git a/packages/rss-handler/src/index.ts b/packages/rss-handler/src/index.ts index b85fe9179..933a9703b 100644 --- a/packages/rss-handler/src/index.ts +++ b/packages/rss-handler/src/index.ts @@ -148,43 +148,43 @@ export const rssHandler = Sentry.GCPFunction.wrapHttpFunction( return res.status(400).send('INVALID_REQUEST_BODY') } - const { userId, feedUrl } = req.body + const { userId, feedUrl, subscriptionId } = req.body // fetch feed const feed = await parser.parseURL(feedUrl) - console.log('Fetched feed', feed.title) + const lastFetchedAt = new Date() + console.log('Fetched feed', feed.title, lastFetchedAt) // save each item in the feed - await Promise.all( - feed.items.map((item) => { - if (!item.link || !item.title || !item.content) { - console.log('Invalid feed item', item) - return - } + for (const item of feed.items) { + if (!item.link || !item.title || !item.content) { + console.log('Invalid feed item', item) + continue + } - const input = { - source: 'rss-feeder', - url: item.link, - saveRequestId: '', - labels: [{ name: 'RSS' }], - title: item.title, - originalContent: item.content, - } + const input = { + source: 'rss-feeder', + url: item.link, + saveRequestId: '', + labels: [{ name: 'RSS' }], + title: item.title, + originalContent: item.content, + } - try { - console.log('Saving page', input.title) - // save page - return sendSavePageMutation(userId, input) - } catch (error) { - console.error('Error while saving page', error) - } - }) - ) + try { + console.log('Saving page', input.title) + // save page + const result = await sendSavePageMutation(userId, input) + console.log('Saved page', result) + } catch (error) { + console.error('Error while saving page', error) + } + } // update subscription lastFetchedAt const updatedSubscription = await sendUpdateSubscriptionMutation( userId, - req.body.subscriptionId, - new Date() + subscriptionId, + lastFetchedAt ) console.log('Updated subscription', updatedSubscription) diff --git a/packages/rss-handler/src/task.ts b/packages/rss-handler/src/task.ts new file mode 100644 index 000000000..94b3352cc --- /dev/null +++ b/packages/rss-handler/src/task.ts @@ -0,0 +1,56 @@ +/* eslint-disable @typescript-eslint/restrict-template-expressions */ +import { CloudTasksClient, protos } from '@google-cloud/tasks' + +const cloudTask = new CloudTasksClient() + +export const emailUserUrl = () => { + const envar = process.env.INTERNAL_SVC_ENDPOINT + if (envar) { + return envar + 'api/user/email' + } + throw 'INTERNAL_SVC_ENDPOINT not set' +} + +export const CONTENT_FETCH_URL = process.env.CONTENT_FETCH_GCF_URL + +export const createCloudTask = async ( + taskHandlerUrl: string | undefined, + payload: unknown, + requestHeaders?: Record, + queue = 'omnivore-import-queue' +) => { + const location = process.env.GCP_LOCATION + const project = process.env.GCP_PROJECT_ID + + if (!project || !location || !queue || !taskHandlerUrl) { + throw `Environment not configured: ${project}, ${location}, ${queue}, ${taskHandlerUrl}` + } + + const serviceAccountEmail = `${project}@appspot.gserviceaccount.com` + + const parent = cloudTask.queuePath(project, location, queue) + const convertedPayload = JSON.stringify(payload) + const body = Buffer.from(convertedPayload).toString('base64') + const task: protos.google.cloud.tasks.v2.ITask = { + httpRequest: { + httpMethod: 'POST', + url: taskHandlerUrl, + headers: { + 'Content-Type': 'application/json', + ...requestHeaders, + }, + body, + ...(serviceAccountEmail + ? { + oidcToken: { + serviceAccountEmail, + }, + } + : null), + }, + } + + return cloudTask.createTask({ parent, task }).then((result) => { + return result[0].name ?? undefined + }) +}