Merge pull request #417 from omnivore-app/feature/upload-highlight-gcs

Upload highlights and labels data to GCS

packages/api/src/datalayer/pubsub.ts

@@ -2,7 +2,6 @@ import { PubSub } from '@google-cloud/pubsub'
 import { env } from '../env'
 import { ReportType } from '../generated/graphql'
 import express from 'express'
-import { Page } from '../elastic/types'
 
 export const createPubSubClient = (): PubsubClient => {
   const client = new PubSub()
@@ -37,17 +36,35 @@ export const createPubSubClient = (): PubsubClient => {
         Buffer.from(JSON.stringify({ userId, email, name, username }))
       )
     },
-    pageUpdated: (page: Partial<Page>, userId: string): Promise<void> => {
+    entityCreated: <T>(
+      type: EntityType,
+      data: T,
+      userId: string
+    ): Promise<void> => {
       return publish(
-        'pageUpdated',
-        Buffer.from(JSON.stringify({ ...page, userId }))
+        'entityCreated',
+        Buffer.from(JSON.stringify({ type, userId, ...data }))
       )
     },
-    pageCreated: (page: Page): Promise<void> => {
-      return publish('pageCreated', Buffer.from(JSON.stringify(page)))
+    entityUpdated: <T>(
+      type: EntityType,
+      data: T,
+      userId: string
+    ): Promise<void> => {
+      return publish(
+        'entityUpdated',
+        Buffer.from(JSON.stringify({ type, userId, ...data }))
+      )
     },
-    pageDeleted: (id: string, userId: string): Promise<void> => {
-      return publish('pageDeleted', Buffer.from(JSON.stringify({ id, userId })))
+    entityDeleted: (
+      type: EntityType,
+      id: string,
+      userId: string
+    ): Promise<void> => {
+      return publish(
+        'entityDeleted',
+        Buffer.from(JSON.stringify({ type, id, userId }))
+      )
     },
     reportSubmitted: (
       submitterId: string,
@@ -65,6 +82,12 @@ export const createPubSubClient = (): PubsubClient => {
   }
 }
 
+export enum EntityType {
+  PAGE = 'page',
+  HIGHLIGHT = 'highlight',
+  LABEL = 'label',
+}
+
 export interface PubsubClient {
   userCreated: (
     userId: string,
@@ -72,9 +95,9 @@ export interface PubsubClient {
     name: string,
     username: string
   ) => Promise<void>
-  pageCreated: (page: Page) => Promise<void>
-  pageUpdated: (page: Partial<Page>, userId: string) => Promise<void>
-  pageDeleted: (id: string, userId: string) => Promise<void>
+  entityCreated: <T>(type: EntityType, data: T, userId: string) => Promise<void>
+  entityUpdated: <T>(type: EntityType, data: T, userId: string) => Promise<void>
+  entityDeleted: (type: EntityType, id: string, userId: string) => Promise<void>
   reportSubmitted(
     submitterId: string | undefined,
     itemUrl: string,

packages/api/src/elastic/highlights.ts

@@ -8,6 +8,7 @@ import {
 import { ResponseError } from '@elastic/elasticsearch/lib/errors'
 import { client, INDEX_ALIAS } from './index'
 import { SortBy, SortOrder, SortParams } from '../utils/search'
+import { EntityType } from '../datalayer/pubsub'
 
 export const addHighlightToPage = async (
   id: string,
@@ -35,7 +36,15 @@ export const addHighlightToPage = async (
       retry_on_conflict: 3,
     })
 
-    return body.result === 'updated'
+    if (body.result !== 'updated') return false
+
+    await ctx.pubsub.entityCreated<Highlight>(
+      EntityType.HIGHLIGHT,
+      highlight,
+      ctx.uid
+    )
+
+    return true
   } catch (e) {
     if (
       e instanceof ResponseError &&
@@ -125,7 +134,11 @@ export const deleteHighlight = async (
       refresh: ctx.refresh,
     })
 
-    return !!body.updated
+    if (body.result !== 'updated') return false
+
+    await ctx.pubsub.entityDeleted(EntityType.HIGHLIGHT, highlightId, ctx.uid)
+
+    return true
   } catch (e) {
     console.error('failed to delete a highlight in elastic', e)
 
@@ -266,7 +279,15 @@ export const updateHighlight = async (
       refresh: ctx.refresh,
     })
 
-    return !!body.updated
+    if (body.result !== 'updated') return false
+
+    await ctx.pubsub.entityUpdated<Highlight>(
+      EntityType.HIGHLIGHT,
+      highlight,
+      ctx.uid
+    )
+
+    return true
   } catch (e) {
     if (
       e instanceof ResponseError &&

packages/api/src/elastic/labels.ts

@@ -1,5 +1,6 @@
 import { Label, PageContext } from './types'
 import { client, INDEX_ALIAS } from './index'
+import { EntityType } from '../datalayer/pubsub'
 
 export const addLabelInPage = async (
   id: string,
@@ -27,9 +28,17 @@ export const addLabelInPage = async (
       retry_on_conflict: 3,
     })
 
-    return body.result === 'updated'
+    if (body.result !== 'updated') return false
+
+    await ctx.pubsub.entityCreated<Label & { pageId: string }>(
+      EntityType.LABEL,
+      { pageId: id, ...label },
+      ctx.uid
+    )
+
+    return true
   } catch (e) {
-    console.error('failed to update a page in elastic', e)
+    console.error('failed to add a label in elastic', e)
     return false
   }
 }
@@ -38,9 +47,9 @@ export const deleteLabelInPages = async (
   userId: string,
   label: string,
   ctx: PageContext
-): Promise<void> => {
+): Promise<boolean> => {
   try {
-    await client.updateByQuery({
+    const { body } = await client.updateByQuery({
       index: INDEX_ALIAS,
       body: {
         script: {
@@ -75,7 +84,14 @@ export const deleteLabelInPages = async (
       },
       refresh: ctx.refresh,
     })
+
+    if (body.result !== 'updated') return false
+
+    await ctx.pubsub.entityDeleted(EntityType.LABEL, label, ctx.uid)
+
+    return true
   } catch (e) {
-    console.error('failed to delete a page in elastic', e)
+    console.error('failed to delete a label in elastic', e)
+    return false
   }
 }

packages/api/src/elastic/pages.ts

@@ -18,6 +18,7 @@ import {
   SortParams,
 } from '../utils/search'
 import { client, INDEX_ALIAS } from './index'
+import { EntityType } from '../datalayer/pubsub'
 
 const appendQuery = (body: SearchBody, query: string): void => {
   body.query.bool.should.push({
@@ -189,7 +190,7 @@ export const createPage = async (
       refresh: ctx.refresh,
     })
 
-    await ctx.pubsub.pageCreated(page)
+    await ctx.pubsub.entityCreated<Page>(EntityType.PAGE, page, ctx.uid)
 
     return body._id as string
   } catch (e) {
@@ -219,7 +220,11 @@ export const updatePage = async (
 
     if (body.result !== 'updated') return false
 
-    await ctx.pubsub.pageUpdated({ ...page, id }, ctx.uid)
+    await ctx.pubsub.entityUpdated<Partial<Page>>(
+      EntityType.PAGE,
+      { ...page, id },
+      ctx.uid
+    )
 
     return true
   } catch (e) {
@@ -241,7 +246,7 @@ export const deletePage = async (
 
     if (body.deleted === 0) return false
 
-    await ctx.pubsub.pageDeleted(id, ctx.uid)
+    await ctx.pubsub.entityDeleted(EntityType.PAGE, id, ctx.uid)
 
     return true
   } catch (e) {

packages/api/src/routers/svc/upload.ts

@@ -6,14 +6,13 @@ import { readPushSubscription } from '../../datalayer/pubsub'
 import { generateUploadSignedUrl, uploadToSignedUrl } from '../../utils/uploads'
 import { v4 as uuidv4 } from 'uuid'
 import { env } from '../../env'
-import { Page } from '../../elastic/types'
 import { DateTime } from 'luxon'
 
-export function pageServiceRouter() {
+export function uploadServiceRouter() {
   const router = express.Router()
 
-  router.post('/upload/:folder', async (req, res) => {
-    console.log('upload page data req', req.params.folder)
+  router.post('/:folder', async (req, res) => {
+    console.log('upload data to folder', req.params.folder)
     const { message: msgStr, expired } = readPushSubscription(req)
 
     if (!msgStr) {
@@ -28,9 +27,9 @@ export function pageServiceRouter() {
     }
 
     try {
-      const data: Partial<Page> = JSON.parse(msgStr)
-      if (!data.userId) {
-        console.log('No userId found in message')
+      const data: { userId: string; type: string } = JSON.parse(msgStr)
+      if (!data.userId || !data.type) {
+        console.log('No userId or type found in message')
         res.status(400).send('Bad Request')
         return
       }
@@ -41,9 +40,9 @@ export function pageServiceRouter() {
       console.log('generate upload url')
 
       const uploadUrl = await generateUploadSignedUrl(
-        `${req.params.folder}/${data.userId}/${DateTime.now().toFormat(
-          'yyyy-LL-dd'
-        )}/${uuidv4()}.json`,
+        `${req.params.folder}/${data.type}/${
+          data.userId
+        }/${DateTime.now().toFormat('yyyy-LL-dd')}/${uuidv4()}.json`,
         contentType,
         bucketName
       )

packages/api/src/server.ts

@@ -40,7 +40,7 @@ import { ApolloServer } from 'apollo-server-express'
 import { pdfAttachmentsRouter } from './routers/svc/pdf_attachments'
 import { corsConfig } from './utils/corsConfig'
 import { initElasticsearch } from './elastic'
-import { pageServiceRouter } from './routers/svc/pages'
+import { uploadServiceRouter } from './routers/svc/upload'
 
 const PORT = process.env.PORT || 4000
 
@@ -98,7 +98,7 @@ export const createApp = (): {
   app.use('/svc/pubsub/links', linkServiceRouter())
   app.use('/svc/pubsub/newsletters', newsletterServiceRouter())
   app.use('/svc/pubsub/emails', emailsServiceRouter())
-  app.use('/svc/pubsub/pages', pageServiceRouter())
+  app.use('/svc/pubsub/upload', uploadServiceRouter())
   app.use('/svc/reminders', remindersServiceRouter())
   app.use('/svc/pdf-attachments', pdfAttachmentsRouter())
 

packages/api/test/routers/pages.test.ts

@@ -1,21 +1,21 @@
 import { request } from '../util'
 import 'mocha'
 
-describe('Pages Router', () => {
+describe('Upload Router', () => {
   const token = process.env.PUBSUB_VERIFICATION_TOKEN || ''
 
   describe('upload', () => {
     it('upload data to GCS', async () => {
       const data = {
         message: {
-          data: Buffer.from(JSON.stringify({ userId: 'userId' })).toString(
+          data: Buffer.from(JSON.stringify({ userId: 'userId', type: 'page' })).toString(
             'base64'
           ),
           publishTime: new Date().toISOString(),
         },
       }
       await request
-        .post(`/svc/pubsub/pages/upload/createdPage?token=${token}`)
+        .post(`/svc/pubsub/upload/createdEntity?token=${token}`)
         .send(data)
         .expect(200)
     })