Send email to user after a bulk import completes

packages/api/src/routers/user_router.ts

@@ -0,0 +1,63 @@
+import express from 'express'
+import { sendEmail } from '../utils/sendEmail'
+import { env } from '../env'
+import { buildLogger } from '../utils/logger'
+import { getRepository } from '../entity/utils'
+import { User } from '../entity/user'
+import { getClaimsByToken } from '../utils/auth'
+
+const logger = buildLogger('app.dispatch')
+
+export function userRouter() {
+  const router = express.Router()
+
+  router.post('/email', async (req, res) => {
+    logger.info('email to-user router')
+
+    const token = req?.cookies?.auth || req?.headers?.authorization
+    const claims = await getClaimsByToken(token)
+    if (!claims) {
+      res.status(401).send('UNAUTHORIZED')
+      return
+    }
+
+    const from = process.env.SENDER_MESSAGE
+    const { body, subject } = req.body as {
+      body?: string
+      subject?: string
+    }
+
+    if (!subject || !body || !from) {
+      console.log(subject, body, from)
+      res.status(400).send('Bad Request')
+      return
+    }
+
+    try {
+      const user = await getRepository(User).findOneBy({ id: claims.uid })
+      if (!user) {
+        res.status(400).send('Bad Request')
+        return
+      }
+
+      const result = await sendEmail({
+        from: env.sender.message,
+        to: user.email,
+        subject: subject,
+        text: body,
+      })
+
+      if (!result) {
+        logger.error('Email not sent to user')
+        res.status(500).send('Failed to send email')
+        return
+      }
+
+      res.status(200).send('Email sent to user')
+    } catch (e) {
+      logger.info(e)
+    }
+  })
+
+  return router
+}

packages/api/src/server.ts

@@ -48,6 +48,7 @@ import { integrationsServiceRouter } from './routers/svc/integrations'
 import { textToSpeechRouter } from './routers/text_to_speech'
 import * as httpContext from 'express-http-context'
 import { notificationRouter } from './routers/notification_router'
+import { userRouter } from './routers/user_router'
 
 const PORT = process.env.PORT || 4000
 
@@ -129,6 +130,7 @@ export const createApp = (): {
 
   app.use('/api/auth', authRouter())
   app.use('/api/page', pageRouter())
+  app.use('/api/user', userRouter())
   app.use('/api/article', articleRouter())
   app.use('/api/mobile-auth', mobileAuthRouter())
   app.use('/api/text-to-speech', textToSpeechRouter())

packages/import-handler/src/index.ts

@@ -10,6 +10,12 @@ import { Stream } from 'node:stream'
 import { v4 as uuid } from 'uuid'
 import { createCloudTask } from './task'
 
+import axios, { AxiosResponse } from 'axios'
+import { promisify } from 'util'
+import * as jwt from 'jsonwebtoken'
+
+const signToken = promisify(jwt.sign)
+
 const storage = new Storage()
 
 interface StorageEventData {
@@ -50,6 +56,30 @@ const importURL = async (
   })
 }
 
+const importCompletedTask = async (userId: string, urlsEnqueued: number) => {
+  if (!process.env.JWT_SECRET) {
+    throw 'Envrionment not setup correctly'
+  }
+
+  const exp = Math.floor(Date.now() / 1000) + 60 * 60 * 24 // 1 day
+  const authToken = await signToken(
+    { uid: userId, exp },
+    process.env.JWT_SECRET
+  )
+  const headers = {
+    Authorization: `auth=${authToken}`,
+  }
+
+  createCloudTask(
+    {
+      userId,
+      subject: 'Your Omnivore import has completed processing',
+      body: `${urlsEnqueued} URLs have been pcoessed and should be available in your library.`,
+    },
+    headers
+  )
+}
+
 const handlerForFile = (name: string): importHandlerFunc | undefined => {
   const fileName = path.parse(name).name
   if (fileName.startsWith('MATTER')) {
@@ -79,21 +109,30 @@ export const importHandler: EventFunction = async (event, context) => {
       return
     }
 
+    const regex = new RegExp('imports/(.*?)/')
+    const groups = regex.exec(data.name)
+    if (!groups || groups.length < 2) {
+      console.log('could not match file pattern: ', data.name)
+      return
+    }
+    const userId = [...groups][1]
+    if (!userId) {
+      console.log('could not extract userId from file name')
+      return
+    }
+
+    let countImported = 0
     await handler(stream, async (url): Promise<void> => {
       try {
         // Imports are stored in the format imports/<user id>/<type>-<uuid>.csv
-        const regex = new RegExp('imports/(.*?)/')
-        const groups = regex.exec(data.name)
-        if (!groups || groups.length < 2) {
-          console.log('could not match file pattern: ', data.name)
-          return
-        }
-        const userId = [...groups][1]
         const result = await importURL(userId, url, 'csv-importer')
         console.log('import url result', result)
+        countImported = countImported + 1
       } catch (err) {
         console.log('error importing url', err)
       }
     })
+
+    await importCompletedTask(userId, countImported)
   }
 }

packages/import-handler/src/task.ts

@@ -10,7 +10,10 @@ type TaskPayload = {
 
 const cloudTask = new CloudTasksClient()
 
-export const createCloudTask = async (payload: TaskPayload) => {
+export const createCloudTask = async (
+  payload: unknown,
+  requestHeaders?: Record<string, string>
+) => {
   const queue = 'omnivore-import-queue'
   const location = process.env.GCP_LOCATION
   const project = process.env.GCP_PROJECT_ID
@@ -40,6 +43,7 @@ export const createCloudTask = async (payload: TaskPayload) => {
       url: taskHandlerUrl,
       headers: {
         'Content-Type': 'application/json',
+        ...requestHeaders,
       },
       body,
       ...(serviceAccountEmail