Save received email before publishing data

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

@@ -99,6 +99,7 @@ export const markEmailAsItemResolver = authorized<
         content: recentEmail.html,
         url: generateUniqueUrl(),
         author: '',
+        receivedEmailId: recentEmail.id,
       },
       newsletterEmail
     )

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

@@ -19,6 +19,9 @@ import {
   saveReceivedEmail,
   updateReceivedEmail,
 } from '../../services/received_emails'
+import cors from 'cors'
+import { corsConfig } from '../../utils/corsConfig'
+import { getClaimsByToken } from '../../utils/auth'
 
 interface EmailMessage {
   from: string
@@ -29,6 +32,7 @@ interface EmailMessage {
   unsubHttpUrl?: string
   text: string
   forwardedFrom?: string
+  receivedEmailId: string
 }
 
 function isEmailMessage(data: any): data is EmailMessage {
@@ -37,7 +41,8 @@ function isEmailMessage(data: any): data is EmailMessage {
     'to' in data &&
     'subject' in data &&
     'html' in data &&
-    'text' in data
+    'text' in data &&
+    'receivedEmailId' in data
   )
 }
 
@@ -98,13 +103,7 @@ export function emailsServiceRouter() {
         })
 
         // update received email type
-        await updateReceivedEmail(
-          user.id,
-          data.from,
-          data.to,
-          data.subject,
-          'article'
-        )
+        await updateReceivedEmail(data.receivedEmailId, 'article')
 
         res.status(200).send('Article')
         return
@@ -146,47 +145,36 @@ export function emailsServiceRouter() {
     }
   })
 
-  // eslint-disable-next-line @typescript-eslint/no-misused-promises
-  router.post('/save', async (req, res) => {
+  router.post('/save', cors<express.Request>(corsConfig), async (req, res) => {
     logger.info('save received email router')
 
-    const { message, expired } = readPushSubscription(req)
-
-    if (!message) {
-      res.status(400).send('Bad Request')
-      return
+    const token = req?.headers?.authorization
+    if (!(await getClaimsByToken(token))) {
+      return res.status(401).send('UNAUTHORIZED')
     }
 
-    if (expired) {
-      logger.info('discards expired message.')
-      res.status(200).send('Expired')
-      return
+    if (!isEmailMessage(req.body)) {
+      logger.error('Invalid message')
+      return res.status(400).send('Bad Request')
     }
 
     try {
-      const data = JSON.parse(message) as unknown
-      if (!isEmailMessage(data)) {
-        logger.error('Invalid message')
-        res.status(400).send('Bad Request')
-        return
-      }
-
       // get user from newsletter email
-      const newsletterEmail = await getNewsletterEmail(data.to)
+      const newsletterEmail = await getNewsletterEmail(req.body.to)
 
       if (!newsletterEmail) {
-        logger.info('newsletter email not found', { email: data.to })
+        logger.info('newsletter email not found', { email: req.body.to })
         res.status(200).send('Not Found')
         return
       }
 
       const user = newsletterEmail.user
-      await saveReceivedEmail(
-        data.from,
-        data.to,
-        data.subject,
-        data.text,
-        data.html,
+      const receivedEmail = await saveReceivedEmail(
+        req.body.from,
+        req.body.to,
+        req.body.subject,
+        req.body.text,
+        req.body.html,
         user.id
       )
 
@@ -198,15 +186,11 @@ export function emailsServiceRouter() {
         },
       })
 
-      res.status(200).send('Received email saved')
+      res.status(200).send({ id: receivedEmail.id })
     } catch (e) {
       logger.info(e)
-      if (e instanceof SyntaxError) {
-        // when message is not a valid json string
-        res.status(400).send(e)
-      } else {
-        res.status(500).send(e)
-      }
+
+      res.status(500).send(e)
     }
   })
 

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

@@ -121,13 +121,7 @@ export function newsletterServiceRouter() {
       }
 
       // update received email type
-      await updateReceivedEmail(
-        newsletterEmail.user.id,
-        data.from,
-        data.email,
-        data.title,
-        'article'
-      )
+      await updateReceivedEmail(data.receivedEmailId, 'article')
 
       // We always send 200 if it was a valid message
       // because we don't want the

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

@@ -88,11 +88,11 @@ export function pdfAttachmentsRouter() {
   router.post('/create-article', async (req, res) => {
     console.log('pdf-attachments/create-article')
 
-    const { email, uploadFileId, subject, from } = req.body as {
+    const { email, uploadFileId, subject, receivedEmailId } = req.body as {
       email: string
       uploadFileId: string
       subject: string
-      from: string
+      receivedEmailId: string
     }
 
     const token = req?.headers?.authorization
@@ -169,13 +169,7 @@ export function pdfAttachmentsRouter() {
       })
 
       // update received email type
-      await updateReceivedEmail(
-        newsletterEmail.user.id,
-        from,
-        email,
-        subject,
-        'article'
-      )
+      await updateReceivedEmail(receivedEmailId, 'article')
 
       res.send({ id: pageId })
     } catch (err) {

packages/api/src/services/received_emails.ts

@@ -22,27 +22,8 @@ export const saveReceivedEmail = async (
 }
 
 export const updateReceivedEmail = async (
-  userId: string,
-  from: string,
-  to: string,
-  subject: string,
+  id: string,
   type: 'article' | 'non-article'
 ) => {
-  await getRepository(ReceivedEmail)
-    .createQueryBuilder()
-    .update()
-    .set({ type })
-    // .where('user_id = :userId', { userId })
-    // .andWhere('from = :from', { from })
-    // .andWhere('to = :to', { to })
-    // .andWhere('subject = :subject', { subject })
-    .where({
-      user: { id: userId },
-      from,
-      to,
-      subject,
-    })
-    .orderBy('created_at', 'DESC')
-    .limit(1)
-    .execute()
+  await getRepository(ReceivedEmail).update(id, { type })
 }

packages/api/src/services/save_newsletter_email.ts

@@ -23,6 +23,7 @@ export interface NewsletterMessage {
   unsubMailTo?: string
   unsubHttpUrl?: string
   text: string
+  receivedEmailId: string
 }
 
 // Returns true if the link was created successfully. Can still fail to

packages/api/test/routers/emails.test.ts

@@ -36,7 +36,7 @@ describe('Emails Router', () => {
     const from = 'from@omnivore.app'
     const to = newsletterEmail
     const subject = 'test subject'
-    const html = 'test html'
+    const html = '<html>test html</html>'
     const text = 'test text'
 
     beforeEach(async () => {

packages/api/test/services/save_newsletter_email.test.ts

@@ -49,6 +49,7 @@ describe('saveNewsletterEmail', () => {
         url,
         title,
         author,
+        receivedEmailId: '',
       },
       newsletterEmail,
       ctx
@@ -83,6 +84,7 @@ describe('saveNewsletterEmail', () => {
         author,
         from: 'fake from',
         text: 'fake text',
+        receivedEmailId: '',
       },
       newsletterEmail,
       ctx

packages/inbound-email-handler/src/index.ts

@@ -14,10 +14,18 @@ import {
 import { PubSub } from '@google-cloud/pubsub'
 import { handlePdfAttachment } from './pdf'
 import { handleNewsletter } from '@omnivore/content-handler'
+import axios from 'axios'
+import { promisify } from 'util'
+import * as jwt from 'jsonwebtoken'
+
+interface SaveReceivedEmailResponse {
+  id: string
+}
+
+const signToken = promisify(jwt.sign)
 
 const NEWSLETTER_EMAIL_RECEIVED_TOPIC = 'newsletterEmailReceived'
 const NON_NEWSLETTER_EMAIL_TOPIC = 'nonNewsletterEmailReceived'
-const RECEIVED_EMAIL_TOPIC = 'receivedEmail'
 const pubsub = new PubSub()
 
 export const publishMessage = async (
@@ -33,8 +41,31 @@ export const publishMessage = async (
     })
 }
 
-const publishReceivedEmail = async (email: any): Promise<void> => {
-  await publishMessage(RECEIVED_EMAIL_TOPIC, email)
+const saveReceivedEmail = async (
+  email: string,
+  data: any
+): Promise<SaveReceivedEmailResponse> => {
+  if (process.env.JWT_SECRET === undefined) {
+    throw new Error('JWT_SECRET is not defined')
+  }
+  const auth = await signToken(email, process.env.JWT_SECRET)
+
+  if (process.env.INTERNAL_SVC_ENDPOINT === undefined) {
+    throw new Error('REST_BACKEND_ENDPOINT is not defined')
+  }
+
+  const response = await axios.post(
+    `${process.env.INTERNAL_SVC_ENDPOINT}svc/pubsub/emails/save`,
+    data,
+    {
+      headers: {
+        Authorization: `${auth as string}`,
+        'Content-Type': 'application/json',
+      },
+    }
+  )
+
+  return response.data as SaveReceivedEmailResponse
 }
 
 export const inboundEmailHandler = Sentry.GCPFunction.wrapHttpFunction(
@@ -81,15 +112,15 @@ export const inboundEmailHandler = Sentry.GCPFunction.wrapHttpFunction(
       const postHeader = headers['list-post']?.toString()
       const unSubHeader = headers['list-unsubscribe']?.toString()
 
-      try {
-        await publishReceivedEmail({
-          from,
-          to,
-          subject,
-          html,
-          text,
-        })
+      const { id: receivedEmailId } = await saveReceivedEmail(to, {
+        from,
+        to,
+        subject,
+        html,
+        text,
+      })
 
+      try {
         // check if it is a confirmation email or forwarding newsletter
         const newsletterMessage = await handleNewsletter({
           from,
@@ -104,6 +135,7 @@ export const inboundEmailHandler = Sentry.GCPFunction.wrapHttpFunction(
             ...newsletterMessage,
             text,
             from,
+            receivedEmailId,
           })
           return res.status(200).send('newsletter received')
         }
@@ -123,7 +155,7 @@ export const inboundEmailHandler = Sentry.GCPFunction.wrapHttpFunction(
             pdfAttachmentName,
             pdfAttachment,
             subject,
-            from
+            receivedEmailId
           )
           return res.send('ok')
         }
@@ -140,6 +172,7 @@ export const inboundEmailHandler = Sentry.GCPFunction.wrapHttpFunction(
             unsubMailTo: unsubscribe.mailTo,
             unsubHttpUrl: unsubscribe.httpUrl,
             forwardedFrom,
+            receivedEmailId,
           },
         })
 
@@ -155,6 +188,7 @@ export const inboundEmailHandler = Sentry.GCPFunction.wrapHttpFunction(
             html,
             text,
             forwardedFrom,
+            receivedEmailId,
           },
         })
       }

packages/inbound-email-handler/src/pdf.ts

@@ -14,7 +14,7 @@ export const handlePdfAttachment = async (
   fileName: string | undefined,
   data: Buffer,
   subject: string,
-  from: string
+  receivedEmailId: string
 ): Promise<void> => {
   console.log('handlePdfAttachment', email, fileName)
 
@@ -27,7 +27,7 @@ export const handlePdfAttachment = async (
       return
     }
     await uploadToSignedUrl(uploadResult.url, data)
-    await createArticle(email, uploadResult.id, subject, from)
+    await createArticle(email, uploadResult.id, subject, receivedEmailId)
   } catch (error) {
     console.error('handlePdfAttachment error', error)
   }
@@ -79,13 +79,13 @@ const createArticle = async (
   email: string,
   uploadFileId: string,
   subject: string,
-  from: string
+  receivedEmailId: string
 ): Promise<AxiosResponse> => {
   const data = {
     email,
     uploadFileId,
     subject,
-    from,
+    receivedEmailId,
   }
 
   if (process.env.JWT_SECRET === undefined) {