Merge pull request #3935 from omnivore-app/feature/get-content-api

feature/get content api

packages/api/src/jobs/find_thumbnail.ts

@@ -127,7 +127,9 @@ export const _findThumbnail = (imagesSizes: (ImageSize | null)[]) => {
 export const findThumbnail = async (data: Data) => {
   const { libraryItemId, userId } = data
 
-  const item = await findLibraryItemById(libraryItemId, userId)
+  const item = await findLibraryItemById(libraryItemId, userId, {
+    select: ['thumbnail', 'readableContent'],
+  })
   if (!item) {
     logger.info('page not found')
     return false

packages/api/src/jobs/save_page.ts

@@ -12,6 +12,7 @@ import { saveFile } from '../services/save_file'
 import { savePage } from '../services/save_page'
 import { uploadFile } from '../services/upload_file'
 import { logError, logger } from '../utils/logger'
+import { downloadFromUrl, uploadToSignedUrl } from '../utils/uploads'
 
 const signToken = promisify(jwt.sign)
 
@@ -47,39 +48,6 @@ const isFetchResult = (obj: unknown): obj is FetchResult => {
   return typeof obj === 'object' && obj !== null && 'finalUrl' in obj
 }
 
-const uploadToSignedUrl = async (
-  uploadSignedUrl: string,
-  contentType: string,
-  contentObjUrl: string
-) => {
-  const maxContentLength = 10 * 1024 * 1024 // 10MB
-
-  logger.info('downloading content', {
-    contentObjUrl,
-  })
-
-  // download the content as stream and max 10MB
-  const response = await axios.get(contentObjUrl, {
-    responseType: 'stream',
-    maxContentLength,
-    timeout: REQUEST_TIMEOUT,
-  })
-
-  logger.info('uploading to signed url', {
-    uploadSignedUrl,
-    contentType,
-  })
-
-  // upload the stream to the signed url
-  await axios.put(uploadSignedUrl, response.data, {
-    headers: {
-      'Content-Type': contentType,
-    },
-    maxBodyLength: maxContentLength,
-    timeout: REQUEST_TIMEOUT,
-  })
-}
-
 const uploadPdf = async (
   url: string,
   userId: string,
@@ -98,7 +66,19 @@ const uploadPdf = async (
     throw new Error('error while getting upload id and signed url')
   }
 
-  await uploadToSignedUrl(result.uploadSignedUrl, 'application/pdf', url)
+  logger.info('downloading content', {
+    url,
+  })
+
+  const data = await downloadFromUrl(url, REQUEST_TIMEOUT)
+
+  const uploadSignedUrl = result.uploadSignedUrl
+  const contentType = 'application/pdf'
+  logger.info('uploading to signed url', {
+    uploadSignedUrl,
+    contentType,
+  })
+  await uploadToSignedUrl(uploadSignedUrl, data, contentType, REQUEST_TIMEOUT)
 
   logger.info('pdf uploaded successfully', {
     url,

packages/api/src/jobs/upload_content.ts

@@ -0,0 +1,65 @@
+import { findLibraryItemById } from '../services/library_item'
+import { logger } from '../utils/logger'
+import { htmlToHighlightedMarkdown, htmlToMarkdown } from '../utils/parser'
+import { uploadToBucket } from '../utils/uploads'
+
+export const UPLOAD_CONTENT_JOB = 'UPLOAD_CONTENT_JOB'
+
+export type ContentFormat = 'markdown' | 'highlightedMarkdown' | 'original'
+
+export interface UploadContentJobData {
+  libraryItemId: string
+  userId: string
+  format: ContentFormat
+  filePath: string
+}
+
+const convertContent = (content: string, format: ContentFormat): string => {
+  switch (format) {
+    case 'markdown':
+      return htmlToMarkdown(content)
+    case 'highlightedMarkdown':
+      return htmlToHighlightedMarkdown(content)
+    case 'original':
+      return content
+    default:
+      throw new Error('Unsupported format')
+  }
+}
+
+const CONTENT_TYPES = {
+  markdown: 'text/markdown',
+  highlightedMarkdown: 'text/markdown',
+  original: 'text/html',
+}
+
+export const uploadContentJob = async (data: UploadContentJobData) => {
+  logger.info('Uploading content to bucket', data)
+
+  const { libraryItemId, userId, format, filePath } = data
+  const libraryItem = await findLibraryItemById(libraryItemId, userId, {
+    select: ['originalContent'],
+  })
+  if (!libraryItem) {
+    logger.error('Library item not found', data)
+    throw new Error('Library item not found')
+  }
+
+  if (!libraryItem.originalContent) {
+    logger.error('Original content not found', data)
+    throw new Error('Original content not found')
+  }
+
+  logger.info('Converting content', data)
+  const content = convertContent(libraryItem.originalContent, format)
+
+  console.time('uploadToBucket')
+  logger.info('Uploading content', data)
+  await uploadToBucket(filePath, Buffer.from(content), {
+    contentType: CONTENT_TYPES[format],
+    timeout: 60000, // 1 minute
+  })
+  console.timeEnd('uploadToBucket')
+
+  logger.info('Content uploaded', data)
+}

packages/api/src/queue-processor.ts

@@ -60,6 +60,7 @@ import {
   UPDATE_LABELS_JOB,
 } from './jobs/update_db'
 import { updatePDFContentJob } from './jobs/update_pdf_content'
+import { uploadContentJob, UPLOAD_CONTENT_JOB } from './jobs/upload_content'
 import { redisDataSource } from './redis_data_source'
 import { CACHED_READING_POSITION_PREFIX } from './services/cached_reading_position'
 import { getJobPriority } from './utils/createTask'
@@ -182,6 +183,8 @@ export const createWorker = (connection: ConnectionOptions) =>
           return forwardEmailJob(job.data)
         case CREATE_DIGEST_JOB:
           return createDigest(job.data)
+        case UPLOAD_CONTENT_JOB:
+          return uploadContentJob(job.data)
         default:
           logger.warning(`[queue-processor] unhandled job: ${job.name}`)
       }

packages/api/src/resolvers/article/index.ts

@@ -399,6 +399,10 @@ export const getArticleResolver = authorized<
           'recommendations.recommender',
           'recommendations_recommender'
         )
+        .leftJoinAndSelect(
+          'recommendations_recommender.profile',
+          'recommendations_recommender_profile'
+        )
         .where('libraryItem.user_id = :uid', { uid })
 
       // We allow the backend to use the ID instead of a slug to fetch the article

packages/api/src/resolvers/article_saving_request/index.ts

@@ -82,7 +82,22 @@ export const articleSavingRequestResolver = authorized<
 
     let libraryItem: LibraryItem | null = null
     if (id) {
-      libraryItem = await findLibraryItemById(id, uid)
+      libraryItem = await findLibraryItemById(id, uid, {
+        select: [
+          'id',
+          'state',
+          'originalUrl',
+          'slug',
+          'title',
+          'author',
+          'createdAt',
+          'updatedAt',
+          'savedAt',
+        ],
+        relations: {
+          user: true,
+        },
+      })
     } else if (url) {
       libraryItem = await findLibraryItemByUrl(cleanUrl(url), uid)
     }

packages/api/src/resolvers/recommendations/index.ts

@@ -141,7 +141,14 @@ export const recommendResolver = authorized<
   MutationRecommendArgs
 >(async (_, { input }, { uid, log, signToken }) => {
   try {
-    const item = await findLibraryItemById(input.pageId, uid)
+    const item = await findLibraryItemById(input.pageId, uid, {
+      select: ['id'],
+      relations: {
+        highlights: {
+          user: true,
+        },
+      },
+    })
     if (!item) {
       return {
         errorCodes: [RecommendErrorCode.NotFound],
@@ -259,7 +266,9 @@ export const recommendHighlightsResolver = authorized<
       }
     }
 
-    const item = await findLibraryItemById(input.pageId, uid)
+    const item = await findLibraryItemById(input.pageId, uid, {
+      select: ['id'],
+    })
     if (!item) {
       return {
         errorCodes: [RecommendHighlightsErrorCode.NotFound],

packages/api/src/routers/article_router.ts

@@ -94,7 +94,9 @@ export function articleRouter() {
       })
 
       try {
-        const item = await findLibraryItemById(articleId, uid)
+        const item = await findLibraryItemById(articleId, uid, {
+          select: ['title', 'readableContent', 'itemLanguage'],
+        })
         if (!item) {
           return res.status(404).send('Page not found')
         }

packages/api/src/routers/content_router.ts

@@ -0,0 +1,125 @@
+import cors from 'cors'
+import express, { Router } from 'express'
+import { ContentFormat, UploadContentJobData } from '../jobs/upload_content'
+import { findLibraryItemsByIds } from '../services/library_item'
+import { getClaimsByToken, getTokenByRequest } from '../utils/auth'
+import { corsConfig } from '../utils/corsConfig'
+import { enqueueBulkUploadContentJob } from '../utils/createTask'
+import { logger } from '../utils/logger'
+import { generateDownloadSignedUrl, isFileExists } from '../utils/uploads'
+
+export function contentRouter() {
+  const router = Router()
+
+  interface GetContentRequest {
+    libraryItemIds: string[]
+    format: ContentFormat
+  }
+
+  const isContentRequest = (data: any): data is GetContentRequest => {
+    return (
+      typeof data === 'object' &&
+      data !== null &&
+      'libraryItemIds' in data &&
+      'format' in data
+    )
+  }
+
+  router.options('/', cors<express.Request>({ ...corsConfig, maxAge: 600 }))
+
+  // eslint-disable-next-line @typescript-eslint/no-misused-promises
+  router.post('/', cors<express.Request>(corsConfig), async (req, res) => {
+    if (!isContentRequest(req.body)) {
+      logger.error('Bad request')
+      return res.status(400).send({ errorCode: 'BAD_REQUEST' })
+    }
+
+    const { libraryItemIds, format } = req.body
+    if (
+      !Array.isArray(libraryItemIds) ||
+      libraryItemIds.length === 0 ||
+      libraryItemIds.length > 50
+    ) {
+      logger.error('Library item ids are invalid')
+      return res.status(400).send({ errorCode: 'BAD_REQUEST' })
+    }
+
+    const token = getTokenByRequest(req)
+    // get claims from token
+    const claims = await getClaimsByToken(token)
+    if (!claims) {
+      logger.error('Token not found')
+      return res.status(401).send({
+        error: 'UNAUTHORIZED',
+      })
+    }
+
+    // get user by uid from claims
+    const userId = claims.uid
+
+    const libraryItems = await findLibraryItemsByIds(libraryItemIds, userId, {
+      select: ['id', 'updatedAt'],
+    })
+    if (libraryItems.length === 0) {
+      logger.error('Library items not found')
+      return res.status(404).send({ errorCode: 'NOT_FOUND' })
+    }
+
+    // generate signed url for each library item
+    const data = await Promise.all(
+      libraryItems.map(async (libraryItem) => {
+        const filePath = `content/${userId}/${
+          libraryItem.id
+        }.${libraryItem.updatedAt.getTime()}.${format}`
+
+        try {
+          const downloadUrl = await generateDownloadSignedUrl(filePath, {
+            expires: Date.now() + 60 * 60 * 1000, // 1 hour
+          })
+
+          // check if file is already uploaded
+          const exists = await isFileExists(filePath)
+          if (exists) {
+            logger.info('File already exists', filePath)
+          }
+
+          return {
+            libraryItemId: libraryItem.id,
+            userId,
+            filePath,
+            downloadUrl,
+            format,
+            exists,
+          }
+        } catch (error) {
+          logger.error('Error while generating signed url', error)
+          return {
+            libraryItemId: libraryItem.id,
+            error: 'Failed to generate download url',
+          }
+        }
+      })
+    )
+    logger.info('Signed urls generated', data)
+
+    // skip uploading if there is an error or file already exists
+    const uploadData = data.filter(
+      (d) => !('error' in d) && d.downloadUrl !== undefined && !d.exists
+    ) as UploadContentJobData[]
+
+    if (uploadData.length > 0) {
+      await enqueueBulkUploadContentJob(uploadData)
+      logger.info('Bulk upload content job enqueued', uploadData)
+    }
+
+    res.send({
+      data: data.map((d) => ({
+        libraryItemId: d.libraryItemId,
+        downloadUrl: d.downloadUrl,
+        error: d.error,
+      })),
+    })
+  })
+
+  return router
+}

packages/api/src/routers/page_router.ts

@@ -146,7 +146,11 @@ export function pageRouter() {
         return res.status(400).send({ errorCode: 'BAD_DATA' })
       }
 
-      const item = await findLibraryItemById(itemId, claims.uid)
+      const item = await findLibraryItemById(itemId, claims.uid, {
+        relations: {
+          highlights: true,
+        },
+      })
       if (!item) {
         return res.status(404).send({ errorCode: 'NOT_FOUND' })
       }

packages/api/src/server.ts

@@ -20,6 +20,7 @@ import { aiSummariesRouter } from './routers/ai_summary_router'
 import { articleRouter } from './routers/article_router'
 import { authRouter } from './routers/auth/auth_router'
 import { mobileAuthRouter } from './routers/auth/mobile/mobile_auth_router'
+import { contentRouter } from './routers/content_router'
 import { digestRouter } from './routers/digest_router'
 import { explainRouter } from './routers/explain_router'
 import { integrationRouter } from './routers/integration_router'
@@ -101,6 +102,8 @@ export const createApp = (): Express => {
   app.use('/api/integration', integrationRouter())
   app.use('/api/tasks', taskRouter())
   app.use('/api/digest', digestRouter())
+  app.use('/api/content', contentRouter())
+
   app.use('/svc/pubsub/content', contentServiceRouter())
   app.use('/svc/pubsub/links', linkServiceRouter())
   app.use('/svc/pubsub/newsletters', newsletterServiceRouter())

packages/api/src/services/library_item.ts

@@ -764,10 +764,18 @@ export const findRecentLibraryItems = async (
   )
 }
 
-export const findLibraryItemsByIds = async (ids: string[], userId: string) => {
-  const selectColumns = getColumns(libraryItemRepository)
-    .filter((column) => column !== 'originalContent')
-    .map((column) => `library_item.${column}`)
+export const findLibraryItemsByIds = async (
+  ids: string[],
+  userId: string,
+  options?: {
+    select?: (keyof LibraryItem)[]
+  }
+) => {
+  const selectColumns =
+    options?.select?.map((column) => `library_item.${column}`) ||
+    getColumns(libraryItemRepository)
+      .filter((column) => column !== 'originalContent')
+      .map((column) => `library_item.${column}`)
   return authTrx(
     async (tx) =>
       tx
@@ -782,17 +790,27 @@ export const findLibraryItemsByIds = async (ids: string[], userId: string) => {
 
 export const findLibraryItemById = async (
   id: string,
-  userId: string
+  userId: string,
+  options?: {
+    select?: (keyof LibraryItem)[]
+    relations?: {
+      user?: boolean
+      labels?: boolean
+      highlights?:
+        | {
+            user?: boolean
+          }
+        | boolean
+    }
+  }
 ): Promise<LibraryItem | null> => {
   return authTrx(
     async (tx) =>
-      tx
-        .createQueryBuilder(LibraryItem, 'library_item')
-        .leftJoinAndSelect('library_item.labels', 'labels')
-        .leftJoinAndSelect('library_item.highlights', 'highlights')
-        .leftJoinAndSelect('highlights.user', 'user')
-        .where('library_item.id = :id', { id })
-        .getOne(),
+      tx.withRepository(libraryItemRepository).findOne({
+        select: options?.select,
+        where: { id },
+        relations: options?.relations,
+      }),
     undefined,
     userId
   )

packages/api/src/services/reports.ts

@@ -11,7 +11,9 @@ export const saveContentDisplayReport = async (
   uid: string,
   input: ReportItemInput
 ): Promise<boolean> => {
-  const item = await findLibraryItemById(input.pageId, uid)
+  const item = await findLibraryItemById(input.pageId, uid, {
+    select: ['id', 'readableContent', 'originalContent', 'originalUrl'],
+  })
   if (!item) {
     logger.info('unable to submit report, item not found', input)
     return false
@@ -53,7 +55,9 @@ export const saveAbuseReport = async (
   uid: string,
   input: ReportItemInput
 ): Promise<boolean> => {
-  const item = await findLibraryItemById(input.pageId, uid)
+  const item = await findLibraryItemById(input.pageId, uid, {
+    select: ['id'],
+  })
   if (!item) {
     logger.info('unable to submit report, item not found', input)
     return false

packages/api/src/utils/createTask.ts

@@ -53,6 +53,10 @@ import {
   UPDATE_HIGHLIGHT_JOB,
   UPDATE_LABELS_JOB,
 } from '../jobs/update_db'
+import {
+  UploadContentJobData,
+  UPLOAD_CONTENT_JOB,
+} from '../jobs/upload_content'
 import { getBackendQueue, JOB_VERSION } from '../queue-processor'
 import { redisDataSource } from '../redis_data_source'
 import { writeDigest } from '../services/digest'
@@ -89,8 +93,9 @@ export const getJobPriority = (jobName: string): number => {
       return 5
     case BULK_ACTION_JOB_NAME:
     case `${REFRESH_FEED_JOB_NAME}_high`:
-      return 10
     case PROCESS_YOUTUBE_TRANSCRIPT_JOB_NAME:
+    case UPLOAD_CONTENT_JOB:
+      return 10
     case `${REFRESH_FEED_JOB_NAME}_low`:
     case EXPORT_ITEM_JOB_NAME:
     case CREATE_DIGEST_JOB:
@@ -953,4 +958,24 @@ export const enqueueCreateDigest = async (
   }
 }
 
+export const enqueueBulkUploadContentJob = async (
+  data: UploadContentJobData[]
+) => {
+  const queue = await getBackendQueue()
+  if (!queue) {
+    return ''
+  }
+
+  const jobs = data.map((d) => ({
+    name: UPLOAD_CONTENT_JOB,
+    data: d,
+    opts: {
+      attempts: 3,
+      priority: getJobPriority(UPLOAD_CONTENT_JOB),
+    },
+  }))
+
+  return queue.addBulk(jobs)
+}
+
 export default createHttpTaskWithToken

packages/api/src/utils/uploads.ts

@@ -1,6 +1,7 @@
 /* eslint-disable @typescript-eslint/no-unsafe-member-access */
 /* eslint-disable @typescript-eslint/no-unsafe-assignment */
 import { File, GetSignedUrlConfig, Storage } from '@google-cloud/storage'
+import axios from 'axios'
 import { ContentReaderType } from '../entity/library_item'
 import { env } from '../env'
 import { PageType } from '../generated/graphql'
@@ -33,6 +34,7 @@ const storage = env.fileUpload?.gcsUploadSAKeyFilePath
   ? new Storage({ keyFilename: env.fileUpload.gcsUploadSAKeyFilePath })
   : new Storage()
 const bucketName = env.fileUpload.gcsUploadBucket
+const maxContentLength = 10 * 1024 * 1024 // 10MB
 
 export const countOfFilesWithPrefix = async (prefix: string) => {
   const [files] = await storage.bucket(bucketName).getFiles({ prefix })
@@ -62,12 +64,16 @@ export const generateUploadSignedUrl = async (
 }
 
 export const generateDownloadSignedUrl = async (
-  filePathName: string
+  filePathName: string,
+  config?: {
+    expires?: number
+  }
 ): Promise<string> => {
   const options: GetSignedUrlConfig = {
     version: 'v4',
     action: 'read',
     expires: Date.now() + 240 * 60 * 1000, // four hours
+    ...config,
   }
   const [url] = await storage
     .bucket(bucketName)
@@ -100,15 +106,50 @@ export const generateUploadFilePathName = (
 export const uploadToBucket = async (
   filePath: string,
   data: Buffer,
-  options?: { contentType?: string; public?: boolean },
+  options?: { contentType?: string; public?: boolean; timeout?: number },
   selectedBucket?: string
 ): Promise<void> => {
   await storage
     .bucket(selectedBucket || bucketName)
     .file(filePath)
-    .save(data, { ...options, timeout: 30000 })
+    .save(data, { timeout: 30000, ...options }) // default timeout 30s
 }
 
 export const createGCSFile = (filename: string): File => {
   return storage.bucket(bucketName).file(filename)
 }
+
+export const downloadFromUrl = async (
+  contentObjUrl: string,
+  timeout?: number
+) => {
+  // download the content as stream and max 10MB
+  const response = await axios.get<Buffer>(contentObjUrl, {
+    responseType: 'stream',
+    maxContentLength,
+    timeout,
+  })
+
+  return response.data
+}
+
+export const uploadToSignedUrl = async (
+  uploadSignedUrl: string,
+  data: Buffer,
+  contentType: string,
+  timeout?: number
+) => {
+  // upload the stream to the signed url
+  await axios.put(uploadSignedUrl, data, {
+    headers: {
+      'Content-Type': contentType,
+    },
+    maxBodyLength: maxContentLength,
+    timeout,
+  })
+}
+
+export const isFileExists = async (filePath: string): Promise<boolean> => {
+  const [exists] = await storage.bucket(bucketName).file(filePath).exists()
+  return exists
+}

packages/api/test/resolvers/article.test.ts

@@ -2345,7 +2345,11 @@ describe('Article API', () => {
         authToken
       ).expect(200)
 
-      const item = await findLibraryItemById(articleId, user.id)
+      const item = await findLibraryItemById(articleId, user.id, {
+        relations: {
+          labels: true,
+        },
+      })
       expect(item?.labels?.map((l) => l.name)).to.eql(['Favorites'])
     })
   })

packages/api/test/resolvers/labels.test.ts

@@ -293,7 +293,11 @@ describe('Labels API', () => {
             labelId,
           }).expect(200)
 
-          const updatedItem = await findLibraryItemById(item.id, user.id)
+          const updatedItem = await findLibraryItemById(item.id, user.id, {
+            relations: {
+              labels: true,
+            },
+          })
           expect(updatedItem?.labels).not.deep.include(toDeleteLabel)
         })
       })
@@ -545,7 +549,11 @@ describe('Labels API', () => {
         it('should update the item with the label', async () => {
           await graphqlRequest(query, authToken).expect(200)
 
-          const updatedItem = await findLibraryItemById(item.id, user.id)
+          const updatedItem = await findLibraryItemById(item.id, user.id, {
+            relations: {
+              labels: true,
+            },
+          })
           const updatedLabel = updatedItem?.labels?.filter(
             (l) => l.id === labelId
           )?.[0]