Add SNS Endpoints for Mail Watcher

packages/local-mail-watcher/src/env.ts

@@ -11,6 +11,9 @@ interface WatcherEnv {
     mq: redisConfig
     cache: redisConfig
   }
+  sns: {
+    snsArn: string
+  }
   apiKey: string
   apiEndpoint: string
 }
@@ -47,10 +50,14 @@ export function getEnv(): WatcherEnv {
       cert: parse('REDIS_CERT')?.replace(/\\n/g, '\n'), // replace \n with new line
     },
   }
+  const sns = {
+    snsArn: parse('SNS_ARN') || '',
+  }
 
   return {
     apiKey: parse('WATCHER_API_KEY')!,
     apiEndpoint: parse('WATCHER_API_ENDPOINT')!,
+    sns,
     filesystem,
     redis
   }

packages/local-mail-watcher/src/index.ts

@@ -7,6 +7,10 @@ import express, {
 
 import { env } from './env'
 import { getQueue } from './lib/queue'
+import { SnsMessage } from './types/SNS'
+import { simpleParser } from 'mailparser'
+import axios from 'axios'
+import { convertToMailObject } from './lib/emailApi'
 
 console.log('Starting worker...')
 
@@ -58,6 +62,40 @@ app.get('/_ah/health', (_req: Request, res: Response) => {
 
 app.post('/mail', addEmailEventToQueue)
 
+app.post('/sns', async (req, res) => {
+  const snsMessage = req.body as SnsMessage
+
+  if (snsMessage.TopicArn != env.sns.snsArn) {
+    res.status(401).send()
+    return
+  }
+
+  if (snsMessage.Type == 'SubscriptionConfirmation') {
+    await axios.get(snsMessage.SubscribeURL)
+    res.status(200).send()
+    return
+  }
+
+  if (snsMessage.Type == 'Received') {
+    const mailContent = await simpleParser(snsMessage.content)
+    const mail = convertToMailObject(mailContent)
+
+    await (
+      await queue
+    ).add('save-newsletter', mail, {
+      priority: 1,
+      attempts: 1,
+      delay: 500,
+    })
+    res.sendStatus(200)
+
+    res.status(200).send()
+    return
+  }
+
+  res.status(400).send()
+})
+
 const port = process.env.PORT || 8080
 const server = app.listen(port, () => {
   console.log('Mail Server started')

packages/local-mail-watcher/src/lib/emailApi.ts

@@ -1,6 +1,7 @@
 import { EmailContents } from '../types/EmailContents'
 import axios from 'axios'
 import { env } from '../env'
+import { ParsedMail } from 'mailparser'
 
 export const sendToEmailApi = (data: EmailContents) => {
   return axios.post(env.apiEndpoint, data, {
@@ -11,3 +12,14 @@ export const sendToEmailApi = (data: EmailContents) => {
     timeout: 5000,
   })
 }
+
+export const convertToMailObject = (it: ParsedMail): EmailContents => {
+  return {
+    from: it.from?.value[0]?.address || '',
+    to: (Array.isArray(it.to) ? it.to[0].text : it.to?.text) || '',
+    subject: it.subject || '',
+    html: it.html || '',
+    text: it.text || '',
+    headers: it.headers,
+  }
+}

packages/local-mail-watcher/src/types/SNS.ts

@@ -0,0 +1,6 @@
+export type SnsMessage = {
+  Type: string
+  TopicArn: string
+  SubscribeURL: string
+  content: string
+}

packages/local-mail-watcher/src/watcher.ts

@@ -1,21 +1,14 @@
 import chokidar from 'chokidar'
 import { simpleParser } from 'mailparser'
 import * as fs from 'node:fs'
-import { sendToEmailApi } from './lib/emailApi'
+import { convertToMailObject, sendToEmailApi } from './lib/emailApi'
 import { env } from './env'
 
 chokidar.watch(env.filesystem.filePath).on('add', (path, _event) => {
   console.log(path)
   const contents = fs.readFileSync(path).toString()
   simpleParser(contents)
-    .then((it) => ({
-      from: it.from?.value[0]?.address || '',
-      to: (Array.isArray(it.to) ? it.to[0].text : it.to?.text) || '',
-      subject: it.subject || '',
-      html: it.html || '',
-      text: it.text || '',
-      headers: it.headers,
-    }))
+    .then(convertToMailObject)
     .then(async (emailData) => {
       await sendToEmailApi(emailData)
       console.log('Sent to email API')