From a806f55233e220ffbd70fb3fa36f655ea0f38b8a Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Wed, 13 Apr 2022 21:18:58 +0800
Subject: [PATCH 1/2] upload highlight to gcs

---
 packages/api/src/datalayer/pubsub.ts          | 26 ++++++++++++++++++-
 packages/api/src/elastic/highlights.ts        | 18 ++++++++++---
 .../src/routers/svc/{pages.ts => upload.ts}   |  7 +++--
 packages/api/src/server.ts                    |  4 +--
 4 files changed, 45 insertions(+), 10 deletions(-)
 rename packages/api/src/routers/svc/{pages.ts => upload.ts} (89%)

diff --git a/packages/api/src/datalayer/pubsub.ts b/packages/api/src/datalayer/pubsub.ts
index 690e9376b..ba98f2952 100644
--- a/packages/api/src/datalayer/pubsub.ts
+++ b/packages/api/src/datalayer/pubsub.ts
@@ -2,7 +2,7 @@ import { PubSub } from '@google-cloud/pubsub'
 import { env } from '../env'
 import { ReportType } from '../generated/graphql'
 import express from 'express'
-import { Page } from '../elastic/types'
+import { Highlight, Page } from '../elastic/types'
 
 export const createPubSubClient = (): PubsubClient => {
   const client = new PubSub()
@@ -49,6 +49,24 @@ export const createPubSubClient = (): PubsubClient => {
     pageDeleted: (id: string, userId: string): Promise<void> => {
       return publish('pageDeleted', Buffer.from(JSON.stringify({ id, userId })))
     },
+    highlightCreated: (highlight: Highlight): Promise<void> => {
+      return publish('highlightCreated', Buffer.from(JSON.stringify(highlight)))
+    },
+    highlightUpdated: (
+      highlight: Partial<Highlight>,
+      userId: string
+    ): Promise<void> => {
+      return publish(
+        'highlightUpdated',
+        Buffer.from(JSON.stringify({ ...highlight, userId }))
+      )
+    },
+    highlightDeleted: (id: string, userId: string): Promise<void> => {
+      return publish(
+        'highlightDeleted',
+        Buffer.from(JSON.stringify({ id, userId }))
+      )
+    },
     reportSubmitted: (
       submitterId: string,
       itemUrl: string,
@@ -75,6 +93,12 @@ export interface PubsubClient {
   pageCreated: (page: Page) => Promise<void>
   pageUpdated: (page: Partial<Page>, userId: string) => Promise<void>
   pageDeleted: (id: string, userId: string) => Promise<void>
+  highlightCreated: (highlight: Highlight) => Promise<void>
+  highlightUpdated: (
+    highlight: Partial<Highlight>,
+    userId: string
+  ) => Promise<void>
+  highlightDeleted: (id: string, userId: string) => Promise<void>
   reportSubmitted(
     submitterId: string | undefined,
     itemUrl: string,
diff --git a/packages/api/src/elastic/highlights.ts b/packages/api/src/elastic/highlights.ts
index 5a499237e..7bd6a6adf 100644
--- a/packages/api/src/elastic/highlights.ts
+++ b/packages/api/src/elastic/highlights.ts
@@ -35,7 +35,11 @@ export const addHighlightToPage = async (
       retry_on_conflict: 3,
     })
 
-    return body.result === 'updated'
+    if (body.result !== 'updated') return false
+
+    await ctx.pubsub.highlightCreated(highlight)
+
+    return true
   } catch (e) {
     if (
       e instanceof ResponseError &&
@@ -125,7 +129,11 @@ export const deleteHighlight = async (
       refresh: ctx.refresh,
     })
 
-    return !!body.updated
+    if (body.result !== 'updated') return false
+
+    await ctx.pubsub.highlightDeleted(highlightId, ctx.uid)
+
+    return true
   } catch (e) {
     console.error('failed to delete a highlight in elastic', e)
 
@@ -266,7 +274,11 @@ export const updateHighlight = async (
       refresh: ctx.refresh,
     })
 
-    return !!body.updated
+    if (body.result !== 'updated') return false
+
+    await ctx.pubsub.highlightUpdated(highlight, ctx.uid)
+
+    return true
   } catch (e) {
     if (
       e instanceof ResponseError &&
diff --git a/packages/api/src/routers/svc/pages.ts b/packages/api/src/routers/svc/upload.ts
similarity index 89%
rename from packages/api/src/routers/svc/pages.ts
rename to packages/api/src/routers/svc/upload.ts
index d1e139094..b26be4cad 100644
--- a/packages/api/src/routers/svc/pages.ts
+++ b/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)
+    console.log('upload data req', req.params.folder)
     const { message: msgStr, expired } = readPushSubscription(req)
 
     if (!msgStr) {
@@ -28,7 +27,7 @@ export function pageServiceRouter() {
     }
 
     try {
-      const data: Partial<Page> = JSON.parse(msgStr)
+      const data: { userId: string } = JSON.parse(msgStr)
       if (!data.userId) {
         console.log('No userId found in message')
         res.status(400).send('Bad Request')
diff --git a/packages/api/src/server.ts b/packages/api/src/server.ts
index 27967ce5a..df70f693e 100755
--- a/packages/api/src/server.ts
+++ b/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())
 

From 0b0b521fabc633c549e3aabd8e280c9c93294466 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Wed, 13 Apr 2022 22:21:20 +0800
Subject: [PATCH 2/2] use generic method for uploading

---
 packages/api/src/datalayer/pubsub.ts    | 63 ++++++++++++-------------
 packages/api/src/elastic/highlights.ts  | 15 ++++--
 packages/api/src/elastic/labels.ts      | 26 ++++++++--
 packages/api/src/elastic/pages.ts       | 11 +++--
 packages/api/src/routers/svc/upload.ts  | 16 +++----
 packages/api/test/routers/pages.test.ts |  6 +--
 6 files changed, 83 insertions(+), 54 deletions(-)

diff --git a/packages/api/src/datalayer/pubsub.ts b/packages/api/src/datalayer/pubsub.ts
index ba98f2952..e4f397acd 100644
--- a/packages/api/src/datalayer/pubsub.ts
+++ b/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 { Highlight, Page } from '../elastic/types'
 
 export const createPubSubClient = (): PubsubClient => {
   const client = new PubSub()
@@ -37,34 +36,34 @@ export const createPubSubClient = (): PubsubClient => {
         Buffer.from(JSON.stringify({ userId, email, name, username }))
       )
     },
-    pageUpdated: (page: Partial<Page>, userId: string): Promise<void> => {
-      return publish(
-        'pageUpdated',
-        Buffer.from(JSON.stringify({ ...page, userId }))
-      )
-    },
-    pageCreated: (page: Page): Promise<void> => {
-      return publish('pageCreated', Buffer.from(JSON.stringify(page)))
-    },
-    pageDeleted: (id: string, userId: string): Promise<void> => {
-      return publish('pageDeleted', Buffer.from(JSON.stringify({ id, userId })))
-    },
-    highlightCreated: (highlight: Highlight): Promise<void> => {
-      return publish('highlightCreated', Buffer.from(JSON.stringify(highlight)))
-    },
-    highlightUpdated: (
-      highlight: Partial<Highlight>,
+    entityCreated: <T>(
+      type: EntityType,
+      data: T,
       userId: string
     ): Promise<void> => {
       return publish(
-        'highlightUpdated',
-        Buffer.from(JSON.stringify({ ...highlight, userId }))
+        'entityCreated',
+        Buffer.from(JSON.stringify({ type, userId, ...data }))
       )
     },
-    highlightDeleted: (id: string, userId: string): Promise<void> => {
+    entityUpdated: <T>(
+      type: EntityType,
+      data: T,
+      userId: string
+    ): Promise<void> => {
       return publish(
-        'highlightDeleted',
-        Buffer.from(JSON.stringify({ id, userId }))
+        'entityUpdated',
+        Buffer.from(JSON.stringify({ type, userId, ...data }))
+      )
+    },
+    entityDeleted: (
+      type: EntityType,
+      id: string,
+      userId: string
+    ): Promise<void> => {
+      return publish(
+        'entityDeleted',
+        Buffer.from(JSON.stringify({ type, id, userId }))
       )
     },
     reportSubmitted: (
@@ -83,6 +82,12 @@ export const createPubSubClient = (): PubsubClient => {
   }
 }
 
+export enum EntityType {
+  PAGE = 'page',
+  HIGHLIGHT = 'highlight',
+  LABEL = 'label',
+}
+
 export interface PubsubClient {
   userCreated: (
     userId: string,
@@ -90,15 +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>
-  highlightCreated: (highlight: Highlight) => Promise<void>
-  highlightUpdated: (
-    highlight: Partial<Highlight>,
-    userId: string
-  ) => Promise<void>
-  highlightDeleted: (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,
diff --git a/packages/api/src/elastic/highlights.ts b/packages/api/src/elastic/highlights.ts
index 7bd6a6adf..934563ca3 100644
--- a/packages/api/src/elastic/highlights.ts
+++ b/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,
@@ -37,7 +38,11 @@ export const addHighlightToPage = async (
 
     if (body.result !== 'updated') return false
 
-    await ctx.pubsub.highlightCreated(highlight)
+    await ctx.pubsub.entityCreated<Highlight>(
+      EntityType.HIGHLIGHT,
+      highlight,
+      ctx.uid
+    )
 
     return true
   } catch (e) {
@@ -131,7 +136,7 @@ export const deleteHighlight = async (
 
     if (body.result !== 'updated') return false
 
-    await ctx.pubsub.highlightDeleted(highlightId, ctx.uid)
+    await ctx.pubsub.entityDeleted(EntityType.HIGHLIGHT, highlightId, ctx.uid)
 
     return true
   } catch (e) {
@@ -276,7 +281,11 @@ export const updateHighlight = async (
 
     if (body.result !== 'updated') return false
 
-    await ctx.pubsub.highlightUpdated(highlight, ctx.uid)
+    await ctx.pubsub.entityUpdated<Highlight>(
+      EntityType.HIGHLIGHT,
+      highlight,
+      ctx.uid
+    )
 
     return true
   } catch (e) {
diff --git a/packages/api/src/elastic/labels.ts b/packages/api/src/elastic/labels.ts
index ec1ac5188..2bb00bba8 100644
--- a/packages/api/src/elastic/labels.ts
+++ b/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
   }
 }
diff --git a/packages/api/src/elastic/pages.ts b/packages/api/src/elastic/pages.ts
index 96f1b5d4a..658abffc7 100644
--- a/packages/api/src/elastic/pages.ts
+++ b/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) {
diff --git a/packages/api/src/routers/svc/upload.ts b/packages/api/src/routers/svc/upload.ts
index b26be4cad..ebc935f86 100644
--- a/packages/api/src/routers/svc/upload.ts
+++ b/packages/api/src/routers/svc/upload.ts
@@ -11,8 +11,8 @@ import { DateTime } from 'luxon'
 export function uploadServiceRouter() {
   const router = express.Router()
 
-  router.post('/upload/:folder', async (req, res) => {
-    console.log('upload 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) {
@@ -27,9 +27,9 @@ export function uploadServiceRouter() {
     }
 
     try {
-      const data: { userId: string } = 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
       }
@@ -40,9 +40,9 @@ export function uploadServiceRouter() {
       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
       )
diff --git a/packages/api/test/routers/pages.test.ts b/packages/api/test/routers/pages.test.ts
index 6778322ea..de7be177f 100644
--- a/packages/api/test/routers/pages.test.ts
+++ b/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)
     })