diff --git a/packages/api/src/apollo.ts b/packages/api/src/apollo.ts
index 2559197ba..26d255e5d 100644
--- a/packages/api/src/apollo.ts
+++ b/packages/api/src/apollo.ts
@@ -28,6 +28,11 @@ import { SetClaimsRole } from './utils/dictionary'
 import { logger } from './utils/logger'
 import { ReadingProgressDataSource } from './datasources/reading_progress_data_source'
 import { createPrometheusExporterPlugin } from '@bmatei/apollo-prometheus-exporter'
+import { ApolloServerPlugin } from 'apollo-server-plugin-base'
+import {
+  countDailyServiceUsage,
+  createServiceUsage,
+} from './services/service_usage'
 
 const signToken = promisify(jwt.sign)
 const pubsub = createPubSubClient()
@@ -112,10 +117,59 @@ export function makeApolloServer(app: Express): ApolloServer {
     },
   })
 
+  // enforce usage limits for the API
+  const usageLimitPlugin = (): ApolloServerPlugin<RequestContext> => {
+    // TODO: load the limit from the DB into memory when the server starts
+    // hardcode the limit for now
+    const MAX_SENT_EMAIL_PER_DAY = 3
+
+    return {
+      async requestDidStart(contextValue) {
+        // get graphql query from the request
+        const query = contextValue.request.query
+        // get the user id from the claims
+        const userId = contextValue.context.claims?.uid
+        const action = 'replyToEmail'
+        if (userId && query?.includes(action)) {
+          logger.info('checking usage limit for user', { userId, action })
+          // get the user's email sent count from the DB
+          const emailSentCount = await countDailyServiceUsage(userId, action)
+          if (emailSentCount >= MAX_SENT_EMAIL_PER_DAY) {
+            logger.info('user has reached the daily email limit', {
+              userId,
+              action,
+            })
+            // if the user has reached the limit, throw an error
+            throw new Error('You have reached the daily email limit')
+          }
+        }
+
+        return {
+          // track usage of the API
+          async willSendResponse(requestContext) {
+            // if the request was successful, increment the user's email sent count
+            if (
+              userId &&
+              query?.includes(action) &&
+              !requestContext.response.errors &&
+              !requestContext.response.data?.replyToEmail?.errorCodes
+            ) {
+              logger.info('incrementing usage count for user', {
+                userId,
+                action,
+              })
+              await createServiceUsage(userId, action)
+            }
+          },
+        }
+      },
+    }
+  }
+
   const apollo = new ApolloServer({
     schema: schema,
     context: contextFunc,
-    plugins: [promExporter],
+    plugins: [promExporter, usageLimitPlugin],
     formatError: (err) => {
       logger.info('server error', err)
       Sentry.captureException(err)
@@ -124,6 +178,7 @@ export function makeApolloServer(app: Express): ApolloServer {
     },
     introspection: env.dev.isLocal,
     persistedQueries: false,
+    stopOnTerminationSignals: false, // we handle this ourselves
   })
 
   return apollo
diff --git a/packages/api/src/entity/received_email.ts b/packages/api/src/entity/received_email.ts
index b68289989..cb5e5937c 100644
--- a/packages/api/src/entity/received_email.ts
+++ b/packages/api/src/entity/received_email.ts
@@ -33,6 +33,12 @@ export class ReceivedEmail {
   @Column('text')
   html!: string
 
+  @Column('text')
+  replyTo?: string
+
+  @Column('text')
+  reply?: string
+
   @Column('text')
   type!: 'article' | 'non-article'
 
diff --git a/packages/api/src/entity/service_usage.ts b/packages/api/src/entity/service_usage.ts
new file mode 100644
index 000000000..056a51fc0
--- /dev/null
+++ b/packages/api/src/entity/service_usage.ts
@@ -0,0 +1,25 @@
+import {
+  Column,
+  CreateDateColumn,
+  Entity,
+  JoinColumn,
+  ManyToOne,
+  PrimaryGeneratedColumn,
+} from 'typeorm'
+import { User } from './user'
+
+@Entity('service_usage')
+export class ServiceUsage {
+  @PrimaryGeneratedColumn('uuid')
+  id!: string
+
+  @ManyToOne(() => User)
+  @JoinColumn({ name: 'user_id' })
+  user!: User
+
+  @Column('varchar')
+  action!: string
+
+  @CreateDateColumn({ default: () => 'CURRENT_TIMESTAMP' })
+  createdAt!: Date
+}
diff --git a/packages/api/src/generated/graphql.ts b/packages/api/src/generated/graphql.ts
index 93afb8908..cd4b67b31 100644
--- a/packages/api/src/generated/graphql.ts
+++ b/packages/api/src/generated/graphql.ts
@@ -58,6 +58,13 @@ export type AddPopularReadSuccess = {
   pageId: Scalars['String'];
 };
 
+export enum AllowedReply {
+  Confirm = 'CONFIRM',
+  Okay = 'OKAY',
+  Subscribe = 'SUBSCRIBE',
+  Yes = 'YES'
+}
+
 export type ApiKey = {
   __typename?: 'ApiKey';
   createdAt: Scalars['Date'];
@@ -1616,6 +1623,7 @@ export type Mutation = {
   optInFeature: OptInFeatureResult;
   recommend: RecommendResult;
   recommendHighlights: RecommendHighlightsResult;
+  replyToEmail: ReplyToEmailResult;
   reportItem: ReportItemResult;
   revokeApiKey: RevokeApiKeyResult;
   saveArticleReadingProgress: SaveArticleReadingProgressResult;
@@ -1836,6 +1844,12 @@ export type MutationRecommendHighlightsArgs = {
 };
 
 
+export type MutationReplyToEmailArgs = {
+  recentEmailId: Scalars['ID'];
+  reply: AllowedReply;
+};
+
+
 export type MutationReportItemArgs = {
   input: ReportItemInput;
 };
@@ -2275,6 +2289,8 @@ export type RecentEmail = {
   from: Scalars['String'];
   html?: Maybe<Scalars['String']>;
   id: Scalars['ID'];
+  reply?: Maybe<Scalars['String']>;
+  replyTo?: Maybe<Scalars['String']>;
   subject: Scalars['String'];
   text: Scalars['String'];
   to: Scalars['String'];
@@ -2430,6 +2446,22 @@ export type ReminderSuccess = {
   reminder: Reminder;
 };
 
+export type ReplyToEmailError = {
+  __typename?: 'ReplyToEmailError';
+  errorCodes: Array<ReplyToEmailErrorCode>;
+};
+
+export enum ReplyToEmailErrorCode {
+  Unauthorized = 'UNAUTHORIZED'
+}
+
+export type ReplyToEmailResult = ReplyToEmailError | ReplyToEmailSuccess;
+
+export type ReplyToEmailSuccess = {
+  __typename?: 'ReplyToEmailSuccess';
+  success: Scalars['Boolean'];
+};
+
 export type ReportItemInput = {
   itemUrl: Scalars['String'];
   pageId: Scalars['ID'];
@@ -3908,6 +3940,7 @@ export type ResolversTypes = {
   AddPopularReadErrorCode: AddPopularReadErrorCode;
   AddPopularReadResult: ResolversTypes['AddPopularReadError'] | ResolversTypes['AddPopularReadSuccess'];
   AddPopularReadSuccess: ResolverTypeWrapper<AddPopularReadSuccess>;
+  AllowedReply: AllowedReply;
   ApiKey: ResolverTypeWrapper<ApiKey>;
   ApiKeysError: ResolverTypeWrapper<ApiKeysError>;
   ApiKeysErrorCode: ApiKeysErrorCode;
@@ -4245,6 +4278,10 @@ export type ResolversTypes = {
   ReminderErrorCode: ReminderErrorCode;
   ReminderResult: ResolversTypes['ReminderError'] | ResolversTypes['ReminderSuccess'];
   ReminderSuccess: ResolverTypeWrapper<ReminderSuccess>;
+  ReplyToEmailError: ResolverTypeWrapper<ReplyToEmailError>;
+  ReplyToEmailErrorCode: ReplyToEmailErrorCode;
+  ReplyToEmailResult: ResolversTypes['ReplyToEmailError'] | ResolversTypes['ReplyToEmailSuccess'];
+  ReplyToEmailSuccess: ResolverTypeWrapper<ReplyToEmailSuccess>;
   ReportItemInput: ReportItemInput;
   ReportItemResult: ResolverTypeWrapper<ReportItemResult>;
   ReportType: ReportType;
@@ -4763,6 +4800,9 @@ export type ResolversParentTypes = {
   ReminderError: ReminderError;
   ReminderResult: ResolversParentTypes['ReminderError'] | ResolversParentTypes['ReminderSuccess'];
   ReminderSuccess: ReminderSuccess;
+  ReplyToEmailError: ReplyToEmailError;
+  ReplyToEmailResult: ResolversParentTypes['ReplyToEmailError'] | ResolversParentTypes['ReplyToEmailSuccess'];
+  ReplyToEmailSuccess: ReplyToEmailSuccess;
   ReportItemInput: ReportItemInput;
   ReportItemResult: ReportItemResult;
   RevokeApiKeyError: RevokeApiKeyError;
@@ -6128,6 +6168,7 @@ export type MutationResolvers<ContextType = ResolverContext, ParentType extends
   optInFeature?: Resolver<ResolversTypes['OptInFeatureResult'], ParentType, ContextType, RequireFields<MutationOptInFeatureArgs, 'input'>>;
   recommend?: Resolver<ResolversTypes['RecommendResult'], ParentType, ContextType, RequireFields<MutationRecommendArgs, 'input'>>;
   recommendHighlights?: Resolver<ResolversTypes['RecommendHighlightsResult'], ParentType, ContextType, RequireFields<MutationRecommendHighlightsArgs, 'input'>>;
+  replyToEmail?: Resolver<ResolversTypes['ReplyToEmailResult'], ParentType, ContextType, RequireFields<MutationReplyToEmailArgs, 'recentEmailId' | 'reply'>>;
   reportItem?: Resolver<ResolversTypes['ReportItemResult'], ParentType, ContextType, RequireFields<MutationReportItemArgs, 'input'>>;
   revokeApiKey?: Resolver<ResolversTypes['RevokeApiKeyResult'], ParentType, ContextType, RequireFields<MutationRevokeApiKeyArgs, 'id'>>;
   saveArticleReadingProgress?: Resolver<ResolversTypes['SaveArticleReadingProgressResult'], ParentType, ContextType, RequireFields<MutationSaveArticleReadingProgressArgs, 'input'>>;
@@ -6293,6 +6334,8 @@ export type RecentEmailResolvers<ContextType = ResolverContext, ParentType exten
   from?: Resolver<ResolversTypes['String'], ParentType, ContextType>;
   html?: Resolver<Maybe<ResolversTypes['String']>, ParentType, ContextType>;
   id?: Resolver<ResolversTypes['ID'], ParentType, ContextType>;
+  reply?: Resolver<Maybe<ResolversTypes['String']>, ParentType, ContextType>;
+  replyTo?: Resolver<Maybe<ResolversTypes['String']>, ParentType, ContextType>;
   subject?: Resolver<ResolversTypes['String'], ParentType, ContextType>;
   text?: Resolver<ResolversTypes['String'], ParentType, ContextType>;
   to?: Resolver<ResolversTypes['String'], ParentType, ContextType>;
@@ -6417,6 +6460,20 @@ export type ReminderSuccessResolvers<ContextType = ResolverContext, ParentType e
   __isTypeOf?: IsTypeOfResolverFn<ParentType, ContextType>;
 };
 
+export type ReplyToEmailErrorResolvers<ContextType = ResolverContext, ParentType extends ResolversParentTypes['ReplyToEmailError'] = ResolversParentTypes['ReplyToEmailError']> = {
+  errorCodes?: Resolver<Array<ResolversTypes['ReplyToEmailErrorCode']>, ParentType, ContextType>;
+  __isTypeOf?: IsTypeOfResolverFn<ParentType, ContextType>;
+};
+
+export type ReplyToEmailResultResolvers<ContextType = ResolverContext, ParentType extends ResolversParentTypes['ReplyToEmailResult'] = ResolversParentTypes['ReplyToEmailResult']> = {
+  __resolveType: TypeResolveFn<'ReplyToEmailError' | 'ReplyToEmailSuccess', ParentType, ContextType>;
+};
+
+export type ReplyToEmailSuccessResolvers<ContextType = ResolverContext, ParentType extends ResolversParentTypes['ReplyToEmailSuccess'] = ResolversParentTypes['ReplyToEmailSuccess']> = {
+  success?: Resolver<ResolversTypes['Boolean'], ParentType, ContextType>;
+  __isTypeOf?: IsTypeOfResolverFn<ParentType, ContextType>;
+};
+
 export type ReportItemResultResolvers<ContextType = ResolverContext, ParentType extends ResolversParentTypes['ReportItemResult'] = ResolversParentTypes['ReportItemResult']> = {
   message?: Resolver<ResolversTypes['String'], ParentType, ContextType>;
   __isTypeOf?: IsTypeOfResolverFn<ParentType, ContextType>;
@@ -7488,6 +7545,9 @@ export type Resolvers<ContextType = ResolverContext> = {
   ReminderError?: ReminderErrorResolvers<ContextType>;
   ReminderResult?: ReminderResultResolvers<ContextType>;
   ReminderSuccess?: ReminderSuccessResolvers<ContextType>;
+  ReplyToEmailError?: ReplyToEmailErrorResolvers<ContextType>;
+  ReplyToEmailResult?: ReplyToEmailResultResolvers<ContextType>;
+  ReplyToEmailSuccess?: ReplyToEmailSuccessResolvers<ContextType>;
   ReportItemResult?: ReportItemResultResolvers<ContextType>;
   RevokeApiKeyError?: RevokeApiKeyErrorResolvers<ContextType>;
   RevokeApiKeyResult?: RevokeApiKeyResultResolvers<ContextType>;
diff --git a/packages/api/src/generated/schema.graphql b/packages/api/src/generated/schema.graphql
index ba442c3a7..e48fe3cff 100644
--- a/packages/api/src/generated/schema.graphql
+++ b/packages/api/src/generated/schema.graphql
@@ -37,6 +37,13 @@ type AddPopularReadSuccess {
   pageId: String!
 }
 
+enum AllowedReply {
+  CONFIRM
+  OKAY
+  SUBSCRIBE
+  YES
+}
+
 type ApiKey {
   createdAt: Date!
   expiresAt: Date!
@@ -1454,6 +1461,7 @@ type Mutation {
   optInFeature(input: OptInFeatureInput!): OptInFeatureResult!
   recommend(input: RecommendInput!): RecommendResult!
   recommendHighlights(input: RecommendHighlightsInput!): RecommendHighlightsResult!
+  replyToEmail(recentEmailId: ID!, reply: AllowedReply!): ReplyToEmailResult!
   reportItem(input: ReportItemInput!): ReportItemResult!
   revokeApiKey(id: ID!): RevokeApiKeyResult!
   saveArticleReadingProgress(input: SaveArticleReadingProgressInput!): SaveArticleReadingProgressResult!
@@ -1671,6 +1679,8 @@ type RecentEmail {
   from: String!
   html: String
   id: ID!
+  reply: String
+  replyTo: String
   subject: String!
   text: String!
   to: String!
@@ -1811,6 +1821,20 @@ type ReminderSuccess {
   reminder: Reminder!
 }
 
+type ReplyToEmailError {
+  errorCodes: [ReplyToEmailErrorCode!]!
+}
+
+enum ReplyToEmailErrorCode {
+  UNAUTHORIZED
+}
+
+union ReplyToEmailResult = ReplyToEmailError | ReplyToEmailSuccess
+
+type ReplyToEmailSuccess {
+  success: Boolean!
+}
+
 input ReportItemInput {
   itemUrl: String!
   pageId: ID!
diff --git a/packages/api/src/jobs/email/inbound_emails.ts b/packages/api/src/jobs/email/inbound_emails.ts
new file mode 100644
index 000000000..db8fc02eb
--- /dev/null
+++ b/packages/api/src/jobs/email/inbound_emails.ts
@@ -0,0 +1,306 @@
+import { handleNewsletter } from '@omnivore/content-handler'
+import { Converter } from 'showdown'
+import { ContentReaderType, LibraryItemState } from '../../entity/library_item'
+import { SubscriptionStatus } from '../../entity/subscription'
+import { UploadFile } from '../../entity/upload_file'
+import { env } from '../../env'
+import { PageType, UploadFileStatus } from '../../generated/graphql'
+import { authTrx } from '../../repository'
+import { createOrUpdateLibraryItem } from '../../services/library_item'
+import {
+  findNewsletterEmailByAddress,
+  updateConfirmationCode,
+} from '../../services/newsletters'
+import {
+  saveReceivedEmail,
+  updateReceivedEmail,
+} from '../../services/received_emails'
+import { saveNewsletter } from '../../services/save_newsletter_email'
+import { saveUrlFromEmail } from '../../services/save_url'
+import { getSubscriptionByName } from '../../services/subscriptions'
+import { analytics } from '../../utils/analytics'
+import { enqueueSendEmail } from '../../utils/createTask'
+import { generateSlug, isUrl } from '../../utils/helpers'
+import { logger } from '../../utils/logger'
+import {
+  parseEmailAddress,
+  isProbablyArticle,
+  getTitleFromEmailSubject,
+  generateUniqueUrl,
+} from '../../utils/parser'
+import {
+  generateUploadFilePathName,
+  getStorageFileDetails,
+} from '../../utils/uploads'
+
+interface EmailJobData {
+  from: string
+  to: string
+  subject: string
+  html: string
+  text: string
+  headers: Record<string, string | string[]>
+  unsubMailTo?: string
+  unsubHttpUrl?: string
+  forwardedFrom?: string
+  replyTo?: string
+  confirmationCode?: string
+  uploadFile?: {
+    fileName: string
+    contentType: string
+    id: string
+  }
+}
+
+const converter = new Converter()
+export const FORWARD_EMAIL_JOB = 'forward-email'
+export const SAVE_NEWSLETTER_JOB = 'save-newsletter'
+export const CONFIRM_EMAIL_JOB = 'confirmation-email'
+export const SAVE_ATTACHMENT_JOB = 'save-attachment'
+
+export const plainTextToHtml = (text: string): string => {
+  return converter.makeHtml(text)
+}
+
+export const forwardEmailJob = async (data: EmailJobData) => {
+  const { from, to, subject, html, text, replyTo, forwardedFrom } = data
+
+  // get user from newsletter email
+  const newsletterEmail = await findNewsletterEmailByAddress(to)
+
+  if (!newsletterEmail) {
+    logger.error(`newsletter email not found: ${to}`)
+    return false
+  }
+
+  const user = newsletterEmail.user
+  const parsedFrom = parseEmailAddress(from)
+
+  const { id: receivedEmailId } = await saveReceivedEmail(
+    from,
+    to,
+    subject,
+    text,
+    html,
+    user.id,
+    'non-article',
+    replyTo
+  )
+
+  if (
+    await isProbablyArticle(
+      forwardedFrom || parsedFrom.address || from,
+      subject
+    )
+  ) {
+    logger.info('handling as article')
+    const savedNewsletter = await saveNewsletter(
+      {
+        title: getTitleFromEmailSubject(subject),
+        author: parsedFrom.name || from,
+        url: generateUniqueUrl(),
+        content: html || text,
+        receivedEmailId,
+        email: newsletterEmail.address,
+      },
+      newsletterEmail
+    )
+    if (!savedNewsletter) {
+      logger.error('Failed to save email', { from, to, subject })
+      return false
+    }
+
+    // update received email type
+    await updateReceivedEmail(receivedEmailId, 'article', user.id)
+
+    return true
+  }
+
+  analytics.capture({
+    distinctId: user.id,
+    event: 'non_newsletter_email_received',
+    properties: {
+      env: env.server.apiEnv,
+    },
+  })
+
+  // forward non-newsletter emails to the registered email address
+  const result = await enqueueSendEmail({
+    from: env.sender.message,
+    to: user.email,
+    subject: `Fwd: ${subject}`,
+    html,
+    text,
+    replyTo: replyTo || from,
+  })
+
+  return !!result
+}
+
+export const saveNewsletterJob = async (data: EmailJobData) => {
+  const {
+    from,
+    to,
+    subject,
+    html,
+    text,
+    replyTo,
+    headers,
+    unsubMailTo,
+    unsubHttpUrl,
+  } = data
+
+  // get user from newsletter email
+  const newsletterEmail = await findNewsletterEmailByAddress(to)
+  if (!newsletterEmail) {
+    logger.error(`newsletter email not found: ${to}`)
+
+    return false
+  }
+
+  const user = newsletterEmail.user
+  const { id: receivedEmailId } = await saveReceivedEmail(
+    from,
+    to,
+    subject,
+    text,
+    html,
+    user.id,
+    'article',
+    replyTo
+  )
+
+  if (isUrl(subject)) {
+    // save url if the title is a parsable url
+    const result = await saveUrlFromEmail(
+      subject,
+      receivedEmailId,
+      newsletterEmail.user.id
+    )
+
+    return result
+  }
+
+  // convert text to html if html is not available
+  const content = html || plainTextToHtml(text)
+  const newsletter = await handleNewsletter({
+    from,
+    to,
+    subject,
+    html: content,
+    headers,
+  })
+
+  const parsedFrom = parseEmailAddress(from)
+  const author = parsedFrom.name || from
+
+  // do not subscribe if subscription already exists and is unsubscribed
+  const existingSubscription = await getSubscriptionByName(
+    author,
+    newsletterEmail.user.id
+  )
+  if (existingSubscription?.status === SubscriptionStatus.Unsubscribed) {
+    logger.info(`newsletter already unsubscribed: ${from}`)
+    return false
+  }
+
+  // save newsletter instead
+  const result = await saveNewsletter(
+    {
+      email: newsletterEmail.address,
+      content,
+      url: generateUniqueUrl(),
+      title: subject,
+      author,
+      unsubMailTo,
+      unsubHttpUrl,
+      receivedEmailId,
+      ...newsletter,
+    },
+    newsletterEmail
+  )
+
+  return result
+}
+
+export const saveAttachmentJob = async (data: EmailJobData) => {
+  const { from, to, subject, html, text, replyTo, uploadFile } = data
+
+  // get user from newsletter email
+  const newsletterEmail = await findNewsletterEmailByAddress(to)
+  if (!newsletterEmail) {
+    logger.error(`newsletter email not found: ${to}`)
+
+    return false
+  }
+
+  const user = newsletterEmail.user
+  await saveReceivedEmail(
+    from,
+    to,
+    subject,
+    text,
+    html,
+    user.id,
+    'article',
+    replyTo
+  )
+
+  const uploadFileData = await authTrx(
+    (tx) =>
+      tx.getRepository(UploadFile).save({
+        ...uploadFile,
+        status: UploadFileStatus.Completed,
+        user: { id: user.id },
+      }),
+    undefined,
+    user.id
+  )
+
+  const uploadFileDetails = await getStorageFileDetails(
+    uploadFileData.id,
+    uploadFileData.fileName
+  )
+
+  const uploadFilePathName = generateUploadFilePathName(
+    uploadFileData.id,
+    uploadFileData.fileName
+  )
+
+  const uploadFileUrlOverride = `https://omnivore.app/attachments/${uploadFilePathName}`
+  const uploadFileHash = uploadFileDetails.md5Hash
+  const itemType =
+    uploadFileData.contentType === 'application/pdf'
+      ? PageType.File
+      : PageType.Book
+  const title = subject || uploadFileData.fileName
+  const itemToCreate = {
+    originalUrl: uploadFileUrlOverride,
+    itemType,
+    textContentHash: uploadFileHash,
+    uploadFile: { id: uploadFileData.id },
+    title,
+    readableContent: '',
+    slug: generateSlug(title),
+    state: LibraryItemState.Succeeded,
+    user: { id: user.id },
+    contentReader:
+      itemType === PageType.File
+        ? ContentReaderType.PDF
+        : ContentReaderType.EPUB,
+  }
+
+  await createOrUpdateLibraryItem(itemToCreate, user.id)
+
+  return true
+}
+
+export const confirmEmailJob = async (data: EmailJobData) => {
+  const { confirmationCode, to } = data
+  if (!confirmationCode) {
+    logger.error('confirmation code not provided')
+    return false
+  }
+
+  return updateConfirmationCode(to, confirmationCode)
+}
diff --git a/packages/api/src/jobs/send_email.ts b/packages/api/src/jobs/email/send_email.ts
similarity index 61%
rename from packages/api/src/jobs/send_email.ts
rename to packages/api/src/jobs/email/send_email.ts
index 8831fd970..130e2b14a 100644
--- a/packages/api/src/jobs/send_email.ts
+++ b/packages/api/src/jobs/email/send_email.ts
@@ -1,27 +1,29 @@
-import { env } from '../env'
-import { sendWithMailJet } from '../services/send_emails'
-import { Merge } from '../util'
-import { logger } from '../utils/logger'
-import { sendEmail } from '../utils/sendEmail'
+import { env } from '../../env'
+import { sendWithMailJet } from '../../services/send_emails'
+import { Merge } from '../../util'
+import { logger } from '../../utils/logger'
+import { sendEmail } from '../../utils/sendEmail'
 
 export const SEND_EMAIL_JOB = 'send-email'
 
 type ContentType = { html: string } | { text: string } | { templateId: string }
 export type SendEmailJobData = Merge<
   {
-    emailAddress: string
+    to: string
+    from?: string
     subject?: string
     html?: string
     text?: string
     templateId?: string
     dynamicTemplateData?: Record<string, any>
+    replyTo?: string
   },
   ContentType
 >
 
 export const sendEmailJob = async (data: SendEmailJobData) => {
   if (process.env.USE_MAILJET && data.dynamicTemplateData) {
-    return sendWithMailJet(data.emailAddress, data.dynamicTemplateData.link)
+    return sendWithMailJet(data.to, data.dynamicTemplateData.link)
   }
 
   if (!data.html && !data.text && !data.templateId) {
@@ -31,7 +33,6 @@ export const sendEmailJob = async (data: SendEmailJobData) => {
 
   return sendEmail({
     ...data,
-    from: env.sender.message,
-    to: data.emailAddress,
+    from: data.from || env.sender.message,
   })
 }
diff --git a/packages/api/src/queue-processor.ts b/packages/api/src/queue-processor.ts
index aab69d3dc..136a854f4 100644
--- a/packages/api/src/queue-processor.ts
+++ b/packages/api/src/queue-processor.ts
@@ -36,7 +36,7 @@ import {
 import { refreshAllFeeds } from './jobs/rss/refreshAllFeeds'
 import { refreshFeed } from './jobs/rss/refreshFeed'
 import { savePageJob } from './jobs/save_page'
-import { sendEmailJob, SEND_EMAIL_JOB } from './jobs/send_email'
+import { sendEmailJob, SEND_EMAIL_JOB } from './jobs/email/send_email'
 import {
   syncReadPositionsJob,
   SYNC_READ_POSITIONS_JOB_NAME,
@@ -53,6 +53,16 @@ import { redisDataSource } from './redis_data_source'
 import { CACHED_READING_POSITION_PREFIX } from './services/cached_reading_position'
 import { getJobPriority } from './utils/createTask'
 import { logger } from './utils/logger'
+import {
+  confirmEmailJob,
+  CONFIRM_EMAIL_JOB,
+  forwardEmailJob,
+  FORWARD_EMAIL_JOB,
+  saveAttachmentJob,
+  saveNewsletterJob,
+  SAVE_ATTACHMENT_JOB,
+  SAVE_NEWSLETTER_JOB,
+} from './jobs/email/inbound_emails'
 
 export const QUEUE_NAME = 'omnivore-backend-queue'
 export const JOB_VERSION = 'v001'
@@ -160,6 +170,14 @@ export const createWorker = (connection: ConnectionOptions) =>
           return exportAllItems(job.data)
         case SEND_EMAIL_JOB:
           return sendEmailJob(job.data)
+        case CONFIRM_EMAIL_JOB:
+          return confirmEmailJob(job.data)
+        case SAVE_ATTACHMENT_JOB:
+          return saveAttachmentJob(job.data)
+        case SAVE_NEWSLETTER_JOB:
+          return saveNewsletterJob(job.data)
+        case FORWARD_EMAIL_JOB:
+          return forwardEmailJob(job.data)
         default:
           logger.warning(`[queue-processor] unhandled job: ${job.name}`)
       }
@@ -304,8 +322,19 @@ const main = async () => {
 
   const gracefulShutdown = async (signal: string) => {
     console.log(`[queue-processor]: Received ${signal}, closing server...`)
+    await new Promise<void>((resolve) => {
+      server.close((err) => {
+        console.log('[queue-processor]: Express server closed')
+        if (err) {
+          console.log('[queue-processor]: error stopping server', { err })
+        }
+
+        resolve()
+      })
+    })
     await worker.close()
     await redisDataSource.shutdown()
+    await appDataSource.destroy()
     process.exit(0)
   }
 
diff --git a/packages/api/src/resolvers/function_resolvers.ts b/packages/api/src/resolvers/function_resolvers.ts
index 9a8fcfa58..95bd3e219 100644
--- a/packages/api/src/resolvers/function_resolvers.ts
+++ b/packages/api/src/resolvers/function_resolvers.ts
@@ -150,7 +150,11 @@ import {
   webhookResolver,
   webhooksResolver,
 } from './index'
-import { markEmailAsItemResolver, recentEmailsResolver } from './recent_emails'
+import {
+  markEmailAsItemResolver,
+  recentEmailsResolver,
+  replyToEmailResolver,
+} from './recent_emails'
 import { recentSearchesResolver } from './recent_searches'
 import { WithDataSourcesContext } from './types'
 import { updateEmailResolver } from './user'
@@ -316,6 +320,7 @@ export const functionResolvers = {
     emptyTrash: emptyTrashResolver,
     fetchContent: fetchContentResolver,
     exportToIntegration: exportToIntegrationResolver,
+    replyToEmail: replyToEmailResolver,
   },
   Query: {
     me: getMeUserResolver,
@@ -680,4 +685,5 @@ export const functionResolvers = {
   ...resultResolveTypeResolver('FetchContent'),
   ...resultResolveTypeResolver('Integration'),
   ...resultResolveTypeResolver('ExportToIntegration'),
+  ...resultResolveTypeResolver('ReplyToEmail'),
 }
diff --git a/packages/api/src/resolvers/recent_emails/index.ts b/packages/api/src/resolvers/recent_emails/index.ts
index a4b86a929..b8bbd275c 100644
--- a/packages/api/src/resolvers/recent_emails/index.ts
+++ b/packages/api/src/resolvers/recent_emails/index.ts
@@ -7,12 +7,17 @@ import {
   MarkEmailAsItemErrorCode,
   MarkEmailAsItemSuccess,
   MutationMarkEmailAsItemArgs,
+  MutationReplyToEmailArgs,
   RecentEmailsError,
-  RecentEmailsErrorCode,
   RecentEmailsSuccess,
+  ReplyToEmailError,
+  ReplyToEmailErrorCode,
+  ReplyToEmailSuccess,
 } from '../../generated/graphql'
+import { getRepository } from '../../repository'
 import { updateReceivedEmail } from '../../services/received_emails'
 import { saveNewsletter } from '../../services/save_newsletter_email'
+import { enqueueSendEmail } from '../../utils/createTask'
 import { authorized } from '../../utils/gql-utils'
 import { generateUniqueUrl, parseEmailAddress } from '../../utils/parser'
 import { sendEmail } from '../../utils/sendEmail'
@@ -20,27 +25,19 @@ import { sendEmail } from '../../utils/sendEmail'
 export const recentEmailsResolver = authorized<
   RecentEmailsSuccess,
   RecentEmailsError
->(async (_, __, { authTrx, log, uid }) => {
-  try {
-    const recentEmails = await authTrx((t) =>
-      t.getRepository(ReceivedEmail).find({
-        where: {
-          user: { id: uid },
-        },
-        order: { createdAt: 'DESC' },
-        take: 20,
-      })
-    )
+>(async (_, __, { authTrx, uid }) => {
+  const recentEmails = await authTrx((t) =>
+    t.getRepository(ReceivedEmail).find({
+      where: {
+        user: { id: uid },
+      },
+      order: { createdAt: 'DESC' },
+      take: 20,
+    })
+  )
 
-    return {
-      recentEmails,
-    }
-  } catch (error) {
-    log.error('Error getting recent emails', error)
-
-    return {
-      errorCodes: [RecentEmailsErrorCode.BadRequest],
-    }
+  return {
+    recentEmails,
   }
 })
 
@@ -49,87 +46,117 @@ export const markEmailAsItemResolver = authorized<
   MarkEmailAsItemError,
   MutationMarkEmailAsItemArgs
 >(async (_, { recentEmailId }, { authTrx, uid, log }) => {
-  try {
-    const recentEmail = await authTrx((t) =>
-      t.getRepository(ReceivedEmail).findOneBy({
-        id: recentEmailId,
+  const recentEmail = await authTrx((t) =>
+    t.getRepository(ReceivedEmail).findOneBy({
+      id: recentEmailId,
+      user: { id: uid },
+      type: 'non-article',
+    })
+  )
+  if (!recentEmail) {
+    log.info('no recent email', recentEmailId)
+
+    return {
+      errorCodes: [MarkEmailAsItemErrorCode.Unauthorized],
+    }
+  }
+
+  const newsletterEmail = await authTrx((t) =>
+    t.getRepository(NewsletterEmail).findOne({
+      where: {
         user: { id: uid },
-        type: 'non-article',
-      })
-    )
-    if (!recentEmail) {
-      log.info('no recent email', recentEmailId)
-
-      return {
-        errorCodes: [MarkEmailAsItemErrorCode.Unauthorized],
-      }
-    }
-
-    const newsletterEmail = await authTrx((t) =>
-      t.getRepository(NewsletterEmail).findOne({
-        where: {
-          user: { id: uid },
-          address: ILike(recentEmail.to),
-        },
-        relations: ['user'],
-      })
-    )
-    if (!newsletterEmail) {
-      log.info('no newsletter email for', {
-        id: recentEmail.id,
-        to: recentEmail.to,
-        from: recentEmail.from,
-      })
-
-      return {
-        errorCodes: [MarkEmailAsItemErrorCode.NotFound],
-      }
-    }
-
-    const success = await saveNewsletter(
-      {
-        from: recentEmail.from,
-        email: recentEmail.to,
-        title: recentEmail.subject,
-        content: recentEmail.html,
-        url: generateUniqueUrl(),
-        author: parseEmailAddress(recentEmail.from).name,
-        receivedEmailId: recentEmail.id,
+        address: ILike(recentEmail.to),
       },
-      newsletterEmail
-    )
-    if (!success) {
-      log.info('newsletter not created', recentEmail.id)
-
-      return {
-        errorCodes: [MarkEmailAsItemErrorCode.BadRequest],
-      }
-    }
-
-    // update received email type
-    await updateReceivedEmail(recentEmail.id, 'article', uid)
-
-    const text = `A recent email marked as a library item
-                    by: ${uid}
-                    from: ${recentEmail.from}
-                    subject: ${recentEmail.subject}`
-
-    // email us to let us know that an email failed to parse as an article
-    await sendEmail({
-      to: env.sender.feedback,
-      subject: 'A recent email marked as a library item',
-      text,
-      from: env.sender.message,
+      relations: ['user'],
+    })
+  )
+  if (!newsletterEmail) {
+    log.info('no newsletter email for', {
+      id: recentEmail.id,
+      to: recentEmail.to,
+      from: recentEmail.from,
     })
 
     return {
-      success,
+      errorCodes: [MarkEmailAsItemErrorCode.NotFound],
     }
-  } catch (error) {
-    log.error('Error marking email as item', error)
+  }
+
+  const success = await saveNewsletter(
+    {
+      from: recentEmail.from,
+      email: recentEmail.to,
+      title: recentEmail.subject,
+      content: recentEmail.html,
+      url: generateUniqueUrl(),
+      author: parseEmailAddress(recentEmail.from).name || recentEmail.from,
+      receivedEmailId: recentEmail.id,
+    },
+    newsletterEmail
+  )
+  if (!success) {
+    log.info('newsletter not created', recentEmail.id)
 
     return {
       errorCodes: [MarkEmailAsItemErrorCode.BadRequest],
     }
   }
+
+  // update received email type
+  await updateReceivedEmail(recentEmail.id, 'article', uid)
+
+  const text = `A recent email marked as a library item
+                    by: ${uid}
+                    from: ${recentEmail.from}
+                    subject: ${recentEmail.subject}`
+
+  // email us to let us know that an email failed to parse as an article
+  await sendEmail({
+    to: env.sender.feedback,
+    subject: 'A recent email marked as a library item',
+    text,
+    from: env.sender.message,
+  })
+
+  return {
+    success,
+  }
+})
+
+export const replyToEmailResolver = authorized<
+  ReplyToEmailSuccess,
+  ReplyToEmailError,
+  MutationReplyToEmailArgs
+>(async (_, { recentEmailId, reply }, { uid, log }) => {
+  const repo = getRepository(ReceivedEmail)
+  const recentEmail = await repo.findOneBy({
+    id: recentEmailId,
+    user: { id: uid },
+  })
+
+  if (!recentEmail) {
+    log.info('no recent email', recentEmailId)
+
+    return {
+      errorCodes: [ReplyToEmailErrorCode.Unauthorized],
+    }
+  }
+
+  const result = await enqueueSendEmail({
+    to: recentEmail.replyTo || recentEmail.from, // send to the reply-to address if it exists or the from address
+    subject: 'Re: ' + recentEmail.subject,
+    text: reply,
+    from: recentEmail.to,
+  })
+
+  const success = !!result
+
+  if (success) {
+    // update received email reply
+    await repo.update(recentEmailId, { reply })
+  }
+
+  return {
+    success,
+  }
 })
diff --git a/packages/api/src/routers/svc/emails.ts b/packages/api/src/routers/svc/emails.ts
index 3d5075cd1..47f8f0b77 100644
--- a/packages/api/src/routers/svc/emails.ts
+++ b/packages/api/src/routers/svc/emails.ts
@@ -27,6 +27,7 @@ interface EmailMessage {
   text: string
   forwardedFrom?: string
   receivedEmailId: string
+  replyTo?: string
 }
 
 function isEmailMessage(data: any): data is EmailMessage {
@@ -82,7 +83,7 @@ export function emailsServiceRouter() {
         const savedNewsletter = await saveNewsletter(
           {
             title: getTitleFromEmailSubject(data.subject),
-            author: parsedFrom.name,
+            author: parsedFrom.name || data.from,
             url: generateUniqueUrl(),
             content: data.html || data.text,
             receivedEmailId: data.receivedEmailId,
@@ -165,7 +166,9 @@ export function emailsServiceRouter() {
         req.body.subject,
         req.body.text,
         req.body.html,
-        user.id
+        user.id,
+        'non-article',
+        req.body.replyTo
       )
 
       analytics.capture({
diff --git a/packages/api/src/schema.ts b/packages/api/src/schema.ts
index 6b0c3cb01..acae6f520 100755
--- a/packages/api/src/schema.ts
+++ b/packages/api/src/schema.ts
@@ -2547,6 +2547,8 @@ const schema = gql`
     type: String!
     text: String!
     html: String
+    replyTo: String
+    reply: String
     createdAt: Date!
   }
 
@@ -3066,6 +3068,27 @@ const schema = gql`
     FAILED_TO_CREATE_TASK
   }
 
+  union ReplyToEmailResult = ReplyToEmailSuccess | ReplyToEmailError
+
+  type ReplyToEmailSuccess {
+    success: Boolean!
+  }
+
+  type ReplyToEmailError {
+    errorCodes: [ReplyToEmailErrorCode!]!
+  }
+
+  enum ReplyToEmailErrorCode {
+    UNAUTHORIZED
+  }
+
+  enum AllowedReply {
+    YES
+    OKAY
+    CONFIRM
+    SUBSCRIBE
+  }
+
   # Mutations
   type Mutation {
     googleLogin(input: GoogleLoginInput!): LoginResult!
@@ -3165,6 +3188,7 @@ const schema = gql`
       contentType: String!
     ): UploadImportFileResult!
     markEmailAsItem(recentEmailId: ID!): MarkEmailAsItemResult!
+    replyToEmail(recentEmailId: ID!, reply: AllowedReply!): ReplyToEmailResult!
     bulkAction(
       query: String!
       action: BulkActionType!
diff --git a/packages/api/src/server.ts b/packages/api/src/server.ts
index 473df7a79..2c945581c 100755
--- a/packages/api/src/server.ts
+++ b/packages/api/src/server.ts
@@ -47,11 +47,7 @@ import { apiLimiter, authLimiter } from './utils/rate_limit'
 
 const PORT = process.env.PORT || 4000
 
-export const createApp = (): {
-  app: Express
-  apollo: ApolloServer
-  httpServer: Server
-} => {
+export const createApp = (): Express => {
   const app = express()
 
   // eslint-disable-next-line @typescript-eslint/ban-ts-comment
@@ -136,10 +132,7 @@ export const createApp = (): {
     res.end(await prom.register.metrics())
   })
 
-  const apollo = makeApolloServer(app)
-  const httpServer = createServer(app)
-
-  return { app, apollo, httpServer }
+  return app
 }
 
 const main = async (): Promise<void> => {
@@ -154,19 +147,17 @@ const main = async (): Promise<void> => {
     await redisDataSource.initialize()
   }
 
-  const { app, apollo, httpServer } = createApp()
-
+  const app = createApp()
+  const apollo = makeApolloServer(app)
   await apollo.start()
   apollo.applyMiddleware({ app, path: '/api/graphql', cors: corsConfig })
 
-  if (!env.dev.isLocal) {
-    const mwLogger = loggers.get('express', { levels: config.syslog.levels })
-    const transport = buildLoggerTransport('express')
-    const mw = await lw.express.makeMiddleware(mwLogger, transport)
-    app.use(mw)
-  }
+  const mwLogger = loggers.get('express', { levels: config.syslog.levels })
+  const transport = buildLoggerTransport('express')
+  const mw = await lw.express.makeMiddleware(mwLogger, transport)
+  app.use(mw)
 
-  const listener = httpServer.listen({ port: PORT }, async () => {
+  const listener = app.listen({ port: PORT }, async () => {
     const logger = buildLogger('app.dispatch')
     logger.notice(`🚀 Server ready at ${apollo.graphqlPath}`)
   })
@@ -181,15 +172,14 @@ const main = async (): Promise<void> => {
   listener.timeout = 640 * 1000 // match headersTimeout
 
   const gracefulShutdown = async (signal: string) => {
+    console.log(`[api]: Received ${signal}, closing server...`)
+    await apollo.stop()
+    console.log('[api]: Apollo server stopped')
+
     console.log('[posthog]: flushing events')
     await analytics.shutdownAsync()
     console.log('[posthog]: events flushed')
 
-    console.log(`[api]: Received ${signal}, closing server...`)
-
-    await apollo.stop()
-    console.log('[api]: Apollo server stopped')
-
     await new Promise<void>((resolve) => {
       listener.close((err) => {
         console.log('[api]: Express listener closed')
diff --git a/packages/api/src/services/received_emails.ts b/packages/api/src/services/received_emails.ts
index b1ea031af..003c82d9d 100644
--- a/packages/api/src/services/received_emails.ts
+++ b/packages/api/src/services/received_emails.ts
@@ -8,7 +8,8 @@ export const saveReceivedEmail = async (
   text: string,
   html: string,
   userId: string,
-  type: 'article' | 'non-article' = 'non-article'
+  type: 'article' | 'non-article' = 'non-article',
+  replyTo?: string
 ): Promise<ReceivedEmail> => {
   return authTrx(
     (t) =>
@@ -20,6 +21,7 @@ export const saveReceivedEmail = async (
         html,
         type,
         user: { id: userId },
+        replyTo,
       }),
     undefined,
     userId
diff --git a/packages/api/src/services/save_newsletter_email.ts b/packages/api/src/services/save_newsletter_email.ts
index 08d6fa67b..f11601288 100644
--- a/packages/api/src/services/save_newsletter_email.ts
+++ b/packages/api/src/services/save_newsletter_email.ts
@@ -76,59 +76,5 @@ export const saveNewsletter = async (
     return false
   }
 
-  // sends push notification
-  // const deviceTokens = await getDeviceTokensByUserId(newsletterEmail.user.id)
-  // if (!deviceTokens) {
-  //   logger.info('Device tokens not set:', newsletterEmail.user.id)
-  //   return true
-  // }
-
-  // const multicastMessage = messageForLink(page, deviceTokens)
-  // await sendMulticastPushNotifications(
-  //   newsletterEmail.user.id,
-  //   multicastMessage,
-  //   'newsletter'
-  // )
-
   return true
 }
-
-// const messageForLink = (
-//   link: Page,
-//   deviceTokens: UserDeviceToken[]
-// ): MulticastMessage => {
-//   let title = '📫 - An article was added to your Omnivore Inbox'
-
-//   if (link.author) {
-//     title = `📫 - ${link.author} has published a new article`
-//   }
-
-//   const pushData = !link
-//     ? undefined
-//     : {
-//         link: Buffer.from(
-//           JSON.stringify({
-//             id: link.id,
-//             url: link.url,
-//             slug: link.slug,
-//             title: link.title,
-//             image: link.image,
-//             author: link.author,
-//             isArchived: !!link.archivedAt,
-//             contentReader: ContentReader.Web,
-//             readingProgressPercent: link.readingProgressPercent,
-//             readingProgressAnchorIndex: link.readingProgressAnchorIndex,
-//           })
-//         ).toString('base64'),
-//       }
-
-//   return {
-//     notification: {
-//       title: title,
-//       body: link.title,
-//       imageUrl: link.image || undefined,
-//     },
-//     data: pushData,
-//     tokens: deviceTokens.map((token) => token.token),
-//   }
-// }
diff --git a/packages/api/src/services/send_emails.ts b/packages/api/src/services/send_emails.ts
index 70dfb019f..69a537bdb 100644
--- a/packages/api/src/services/send_emails.ts
+++ b/packages/api/src/services/send_emails.ts
@@ -18,7 +18,7 @@ export const sendNewAccountVerificationEmail = async (user: {
   }
 
   const result = await enqueueSendEmail({
-    emailAddress: user.email,
+    to: user.email,
     dynamicTemplateData: dynamicTemplateData,
     templateId: env.sendgrid.confirmationTemplateId,
   })
@@ -78,7 +78,7 @@ export const sendAccountChangeEmail = async (user: {
   }
 
   const result = await enqueueSendEmail({
-    emailAddress: user.email,
+    to: user.email,
     dynamicTemplateData: dynamicTemplateData,
     templateId: env.sendgrid.verificationTemplateId,
   })
@@ -100,7 +100,7 @@ export const sendPasswordResetEmail = async (user: {
   }
 
   const result = await enqueueSendEmail({
-    emailAddress: user.email,
+    to: user.email,
     dynamicTemplateData: dynamicTemplateData,
     templateId: env.sendgrid.resetPasswordTemplateId,
   })
diff --git a/packages/api/src/services/service_usage.ts b/packages/api/src/services/service_usage.ts
new file mode 100644
index 000000000..454f3d9a9
--- /dev/null
+++ b/packages/api/src/services/service_usage.ts
@@ -0,0 +1,31 @@
+import { Between } from 'typeorm'
+import { ServiceUsage } from '../entity/service_usage'
+import { authTrx, getRepository } from '../repository'
+import { DateTime } from 'luxon'
+
+const repo = getRepository(ServiceUsage)
+
+export const countDailyServiceUsage = async (
+  userId: string,
+  action: string
+) => {
+  return authTrx((tx) =>
+    tx.withRepository(repo).countBy({
+      user: { id: userId },
+      action,
+      createdAt: Between(
+        DateTime.now().startOf('day').toJSDate(),
+        DateTime.now().endOf('day').toJSDate()
+      ),
+    })
+  )
+}
+
+export const createServiceUsage = async (userId: string, action: string) => {
+  return authTrx((tx) =>
+    tx.withRepository(repo).save({
+      user: { id: userId },
+      action,
+    })
+  )
+}
diff --git a/packages/api/src/utils/createTask.ts b/packages/api/src/utils/createTask.ts
index c2ce23fa9..001daee64 100644
--- a/packages/api/src/utils/createTask.ts
+++ b/packages/api/src/utils/createTask.ts
@@ -35,7 +35,7 @@ import {
   REFRESH_ALL_FEEDS_JOB_NAME,
   REFRESH_FEED_JOB_NAME,
 } from '../jobs/rss/refreshAllFeeds'
-import { SendEmailJobData, SEND_EMAIL_JOB } from '../jobs/send_email'
+import { SendEmailJobData, SEND_EMAIL_JOB } from '../jobs/email/send_email'
 import { SYNC_READ_POSITIONS_JOB_NAME } from '../jobs/sync_read_positions'
 import { TriggerRuleJobData, TRIGGER_RULE_JOB_NAME } from '../jobs/trigger_rule'
 import {
diff --git a/packages/api/test/global-teardown.ts b/packages/api/test/global-teardown.ts
index 922d5db46..383fd6e36 100644
--- a/packages/api/test/global-teardown.ts
+++ b/packages/api/test/global-teardown.ts
@@ -7,16 +7,16 @@ export const mochaGlobalTeardown = async () => {
   await stopApolloServer()
   console.log('apollo server stopped')
 
-  await appDataSource.destroy()
-  console.log('db connection closed')
-
   if (env.redis.cache.url) {
-    await redisDataSource.shutdown()
-    console.log('redis connection closed')
-
     if (redisDataSource.workerRedisClient) {
       await stopWorker()
       console.log('worker closed')
     }
+
+    await redisDataSource.shutdown()
+    console.log('redis connection closed')
   }
+
+  await appDataSource.destroy()
+  console.log('db connection closed')
 }
diff --git a/packages/api/test/util.ts b/packages/api/test/util.ts
index 6717fdbe1..ee1c30bdc 100644
--- a/packages/api/test/util.ts
+++ b/packages/api/test/util.ts
@@ -2,11 +2,13 @@ import { ConnectionOptions, Job, QueueEvents, Worker } from 'bullmq'
 import { nanoid } from 'nanoid'
 import supertest from 'supertest'
 import { v4 } from 'uuid'
+import { makeApolloServer } from '../src/apollo'
 import { createWorker, QUEUE_NAME } from '../src/queue-processor'
 import { createApp } from '../src/server'
 import { corsConfig } from '../src/utils/corsConfig'
 
-const { app, apollo } = createApp()
+const app = createApp()
+const apollo = makeApolloServer(app)
 export const request = supertest(app)
 let worker: Worker
 let queueEvents: QueueEvents
diff --git a/packages/db/migrations/0172.do.service_usage.sql b/packages/db/migrations/0172.do.service_usage.sql
new file mode 100755
index 000000000..fa02e6214
--- /dev/null
+++ b/packages/db/migrations/0172.do.service_usage.sql
@@ -0,0 +1,39 @@
+-- Type: DO
+-- Name: service_usage
+-- Description: Create table for tracking service usage and enforce limit
+
+BEGIN;
+
+ALTER TABLE omnivore.received_emails
+    ADD COLUMN reply_to TEXT,
+    ADD COLUMN reply TEXT;
+
+CREATE TABLE omnivore.subscription_plan (
+    id SERIAL PRIMARY KEY,
+    name VARCHAR(255) NOT NULL,
+    description TEXT,
+    max_emails_sent_per_day INT NOT NULL,
+    created_at timestamptz NOT NULL default current_timestamp
+);
+
+INSERT INTO omnivore.subscription_plan (id, name, description, max_emails_sent_per_day)
+VALUES (1, 'Basic', 'Basic plan', 3);
+
+CREATE TABLE omnivore.service_usage (
+	id uuid PRIMARY KEY DEFAULT uuid_generate_v1mc(),
+    user_id uuid NOT NULL REFERENCES omnivore.user,
+    action VARCHAR(255) NOT NULL,
+    created_at timestamptz NOT NULL default current_timestamp
+);
+
+CREATE INDEX ON omnivore.service_usage (user_id);
+
+ALTER TABLE omnivore.service_usage ENABLE ROW LEVEL SECURITY;
+
+CREATE POLICY service_usage_policy on omnivore.service_usage
+    USING (user_id = omnivore.get_current_user_id())
+    WITH CHECK (user_id = omnivore.get_current_user_id());
+
+GRANT SELECT, INSERT ON omnivore.service_usage TO omnivore_user;
+
+COMMIT;
diff --git a/packages/db/migrations/0172.undo.service_usage.sql b/packages/db/migrations/0172.undo.service_usage.sql
new file mode 100755
index 000000000..99d324323
--- /dev/null
+++ b/packages/db/migrations/0172.undo.service_usage.sql
@@ -0,0 +1,15 @@
+-- Type: UNDO
+-- Name: service_usage
+-- Description: Create table for tracking service usage and enforce limit
+
+BEGIN;
+
+DROP TABLE IF EXISTS omnivore.service_usage;
+
+DROP TABLE IF EXISTS omnivore.subscription_plan;
+
+ATLER TABLE omnivore.received_emails
+    DROP COLUMN IF EXISTS reply_to,
+    DROP COLUMN IF EXISTS reply;
+
+COMMIT;
diff --git a/packages/inbound-email-handler/.eslintrc b/packages/inbound-email-handler/.eslintrc
index e006282a6..301be9795 100644
--- a/packages/inbound-email-handler/.eslintrc
+++ b/packages/inbound-email-handler/.eslintrc
@@ -2,5 +2,13 @@
   "extends": "../../.eslintrc",
   "parserOptions": {
     "project": "tsconfig.json"
+  },
+  "rules": {
+    "@typescript-eslint/no-floating-promises": [
+      "error",
+      {
+        "ignoreIIFE": true
+      }
+    ]
   }
-}
\ No newline at end of file
+}
diff --git a/packages/inbound-email-handler/mocha-config.json b/packages/inbound-email-handler/mocha-config.json
index 5a2631f42..897a3bb1e 100644
--- a/packages/inbound-email-handler/mocha-config.json
+++ b/packages/inbound-email-handler/mocha-config.json
@@ -1,4 +1,6 @@
 {
     "extension": ["ts"],
-    "spec": "test/**/*.test.ts"
+    "spec": "test/**/*.test.ts",
+    "require": ["test/global-teardown.ts"],
+    "timeout": 10000
   }
diff --git a/packages/inbound-email-handler/package.json b/packages/inbound-email-handler/package.json
index 2330a4308..2581a61a2 100644
--- a/packages/inbound-email-handler/package.json
+++ b/packages/inbound-email-handler/package.json
@@ -22,7 +22,9 @@
   },
   "devDependencies": {
     "@types/addressparser": "^1.0.1",
+    "@types/chai": "^4.3.6",
     "@types/json-bigint": "^1.0.1",
+    "@types/mocha": "^10.0.0",
     "@types/node": "^14.11.2",
     "@types/rfc2047": "^2.0.1",
     "@types/showdown": "^2.0.1",
@@ -32,17 +34,16 @@
   },
   "dependencies": {
     "@google-cloud/functions-framework": "3.1.2",
-    "@google-cloud/pubsub": "^4.0.0",
-    "@omnivore/content-handler": "1.0.0",
-    "@sendgrid/client": "^7.6.0",
+    "@google-cloud/storage": "^7.0.1",
     "@sentry/serverless": "^7.77.0",
     "addressparser": "^1.0.1",
-    "axios": "^0.27.2",
-    "jsonwebtoken": "^8.5.1",
+    "bullmq": "^5.1.1",
+    "dotenv": "^8.2.0",
+    "ioredis": "^5.3.2",
     "parse-headers": "^2.0.4",
     "parse-multipart-data": "^1.2.1",
     "rfc2047": "^4.0.1",
-    "showdown": "^2.1.0"
+    "uuid": "^8.3.1"
   },
   "volta": {
     "extends": "../../package.json"
diff --git a/packages/inbound-email-handler/src/attachment.ts b/packages/inbound-email-handler/src/attachment.ts
index 273555bae..db3d0d094 100644
--- a/packages/inbound-email-handler/src/attachment.ts
+++ b/packages/inbound-email-handler/src/attachment.ts
@@ -1,8 +1,11 @@
-import axios, { AxiosResponse } from 'axios'
-import * as jwt from 'jsonwebtoken'
-import { promisify } from 'util'
+import { Storage } from '@google-cloud/storage'
+import { v4 as uuid } from 'uuid'
+import { EmailJobType, queueEmailJob } from './job'
 
-const signToken = promisify(jwt.sign)
+const storage = process.env.GCS_UPLOAD_SA_KEY_FILE_PATH
+  ? new Storage({ keyFilename: process.env.GCS_UPLOAD_SA_KEY_FILE_PATH })
+  : new Storage()
+const bucketName = process.env.GCS_UPLOAD_BUCKET || 'omnivore-files'
 
 export interface Attachment {
   contentType: string
@@ -10,11 +13,6 @@ export interface Attachment {
   filename: string | undefined
 }
 
-type UploadResponse = {
-  id: string
-  url: string
-}
-
 export const isAttachment = (contentType: string, data: Buffer): boolean => {
   return (
     (contentType === 'application/pdf' ||
@@ -23,11 +21,26 @@ export const isAttachment = (contentType: string, data: Buffer): boolean => {
   )
 }
 
+export const uploadToBucket = async (
+  fileName: string,
+  data: Buffer,
+  options?: { contentType?: string; public?: boolean }
+) => {
+  const uploadFileId = uuid()
+
+  await storage
+    .bucket(bucketName)
+    .file(`u/${uploadFileId}/${fileName}`)
+    .save(data, { ...options, timeout: 30000 })
+
+  return uploadFileId
+}
+
 export const handleAttachments = async (
-  email: string,
+  from: string,
+  to: string,
   subject: string,
-  attachments: Attachment[],
-  receivedEmailId: string
+  attachments: Attachment[]
 ): Promise<void> => {
   for await (const attachment of attachments) {
     const { contentType, data } = attachment
@@ -36,98 +49,20 @@ export const handleAttachments = async (
         ? 'attachment.pdf'
         : 'attachment.epub'
 
-    try {
-      const uploadResult = await getUploadIdAndSignedUrl(
-        email,
-        filename,
-        contentType
-      )
-      if (!uploadResult.url || !uploadResult.id) {
-        console.log('failed to create upload request', uploadResult)
-        return
-      }
-      await uploadToSignedUrl(uploadResult.url, data, contentType)
-      await createArticle(email, uploadResult.id, subject, receivedEmailId)
-    } catch (error) {
-      console.error('handleAttachments error', error)
-    }
-  }
-}
+    const uploadFileId = await uploadToBucket(filename, data, {
+      contentType,
+      public: false,
+    })
 
-const getUploadIdAndSignedUrl = async (
-  email: string,
-  fileName: string,
-  contentType: string
-): Promise<UploadResponse> => {
-  if (process.env.JWT_SECRET === undefined) {
-    throw new Error('JWT_SECRET is not defined')
-  }
-  const auth = await signToken(email, process.env.JWT_SECRET)
-  const data = {
-    fileName,
-    email,
-    contentType,
-  }
-
-  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/email-attachment/upload`,
-    data,
-    {
-      headers: {
-        Authorization: `${auth as string}`,
-        'Content-Type': 'application/json',
+    await queueEmailJob(EmailJobType.SaveAttachment, {
+      from,
+      to,
+      uploadFile: {
+        fileName: filename,
+        contentType,
+        id: uploadFileId,
       },
-    }
-  )
-  return response.data as UploadResponse
-}
-
-const uploadToSignedUrl = async (
-  uploadUrl: string,
-  data: Buffer,
-  contentType: string
-): Promise<AxiosResponse> => {
-  return axios.put(uploadUrl, data, {
-    headers: {
-      'Content-Type': contentType,
-    },
-    maxBodyLength: 1000000000,
-    maxContentLength: 100000000,
-  })
-}
-
-const createArticle = async (
-  email: string,
-  uploadFileId: string,
-  subject: string,
-  receivedEmailId: string
-): Promise<AxiosResponse> => {
-  const data = {
-    email,
-    uploadFileId,
-    subject,
-    receivedEmailId,
+      subject,
+    })
   }
-
-  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')
-  }
-  return axios.post(
-    `${process.env.INTERNAL_SVC_ENDPOINT}svc/email-attachment/create-article`,
-    data,
-    {
-      headers: {
-        Authorization: `${auth as string}`,
-        'Content-Type': 'application/json',
-      },
-    }
-  )
 }
diff --git a/packages/inbound-email-handler/src/index.ts b/packages/inbound-email-handler/src/index.ts
index eefadb6a3..0abd80dac 100644
--- a/packages/inbound-email-handler/src/index.ts
+++ b/packages/inbound-email-handler/src/index.ts
@@ -2,92 +2,29 @@
 /* eslint-disable @typescript-eslint/no-explicit-any */
 /* eslint-disable @typescript-eslint/no-unsafe-argument */
 /* eslint-disable @typescript-eslint/no-unused-vars */
-
-import { PubSub } from '@google-cloud/pubsub'
-import { handleNewsletter } from '@omnivore/content-handler'
-import { generateUniqueUrl } from '@omnivore/content-handler/build/src/content-handler'
 import * as Sentry from '@sentry/serverless'
-import axios from 'axios'
-import * as jwt from 'jsonwebtoken'
 import parseHeaders from 'parse-headers'
 import * as multipart from 'parse-multipart-data'
 import rfc2047 from 'rfc2047'
-import { Converter } from 'showdown'
-import { promisify } from 'util'
 import { Attachment, handleAttachments, isAttachment } from './attachment'
+import { EmailJobType, queueEmailJob } from './job'
 import {
   handleGoogleConfirmationEmail,
   isGoogleConfirmationEmail,
   isSubscriptionConfirmationEmail,
-  parseAuthor,
   parseUnsubscribe,
 } from './newsletter'
 
-interface SaveReceivedEmailResponse {
-  id: string
-}
-
 interface Envelope {
   to: string[]
   from: string
 }
 
-const signToken = promisify(jwt.sign)
-
 Sentry.GCPFunction.init({
   dsn: process.env.SENTRY_DSN,
   tracesSampleRate: 0,
 })
 
-const NEWSLETTER_EMAIL_RECEIVED_TOPIC = 'newsletterEmailReceived'
-const NON_NEWSLETTER_EMAIL_TOPIC = 'nonNewsletterEmailReceived'
-const pubsub = new PubSub()
-const converter = new Converter()
-
-export const plainTextToHtml = (text: string): string => {
-  return converter.makeHtml(text)
-}
-
-export const publishMessage = async (
-  topic: string,
-  message: any
-): Promise<string | undefined> => {
-  return pubsub
-    .topic(topic)
-    .publishMessage({ json: message })
-    .catch((err) => {
-      console.log('error publishing message:', err)
-      return undefined
-    })
-}
-
-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 parsedTo = (parsed: Record<string, string>): string => {
   // envelope to contains the real recipient email address
   try {
@@ -120,6 +57,7 @@ export const inboundEmailHandler = Sentry.GCPFunction.wrapHttpFunction(
 
       // original sender email address
       const from = parsed['from']
+      const replyTo = parsed['reply-to']
       const subject = parsed['subject']
       const html = parsed['html']
       const text = parsed['text']
@@ -134,96 +72,77 @@ export const inboundEmailHandler = Sentry.GCPFunction.wrapHttpFunction(
         ? parseUnsubscribe(unSubHeader)
         : undefined
 
-      const { id: receivedEmailId } = await saveReceivedEmail(to, {
-        from,
-        to,
-        subject,
-        html,
-        text,
-      })
-
       try {
         // check if it is a subscription or google confirmation email
         const isGoogleConfirmation = isGoogleConfirmationEmail(from, subject)
         if (isGoogleConfirmation || isSubscriptionConfirmationEmail(subject)) {
           console.debug('handleConfirmation', from, subject)
           // we need to parse the confirmation code from the email
-          isGoogleConfirmation &&
-            (await handleGoogleConfirmationEmail(to, subject))
-          // queue non-newsletter emails
-          await pubsub.topic(NON_NEWSLETTER_EMAIL_TOPIC).publishMessage({
-            json: {
-              from,
-              to,
-              subject,
-              html,
-              text,
-              unsubMailTo: unsubscribe?.mailTo,
-              unsubHttpUrl: unsubscribe?.httpUrl,
-              forwardedFrom,
-              receivedEmailId,
-            },
+          if (isGoogleConfirmation) {
+            await handleGoogleConfirmationEmail(from, to, subject)
+          }
+
+          // forward emails
+          await queueEmailJob(EmailJobType.ForwardEmail, {
+            from,
+            to,
+            subject,
+            html,
+            text,
+            headers,
+            forwardedFrom,
+            replyTo,
           })
           return res.send('ok')
         }
         if (attachments.length > 0) {
           console.debug('handle attachments', from, to, subject)
           // save the attachments as articles
-          await handleAttachments(to, subject, attachments, receivedEmailId)
+          await handleAttachments(from, to, subject, attachments)
           return res.send('ok')
         }
 
-        // convert text to html if html is not available
-        const content = html || plainTextToHtml(text)
-
         // all other emails are considered newsletters
-        const newsletterMessage = await handleNewsletter({
+        // queue newsletter emails
+        await queueEmailJob(EmailJobType.SaveNewsletter, {
           from,
           to,
           subject,
-          html: content,
+          html,
+          text,
           headers,
+          unsubMailTo: unsubscribe?.mailTo,
+          unsubHttpUrl: unsubscribe?.httpUrl,
+          forwardedFrom,
+          replyTo,
         })
 
-        // queue newsletter emails
-        await pubsub.topic(NEWSLETTER_EMAIL_RECEIVED_TOPIC).publishMessage({
-          json: {
-            email: to,
-            content,
-            url: generateUniqueUrl(),
-            title: subject,
-            author: parseAuthor(from),
-            unsubMailTo: unsubscribe?.mailTo,
-            unsubHttpUrl: unsubscribe?.httpUrl,
-            receivedEmailId,
-            ...newsletterMessage,
-          },
-        })
         res.send('newsletter received')
       } catch (error) {
-        console.log(
+        console.error(
           'error handling emails, will forward.',
           from,
           to,
           subject,
           error
         )
-        // queue error emails
-        await pubsub.topic(NON_NEWSLETTER_EMAIL_TOPIC).publishMessage({
-          json: {
-            from,
-            to,
-            subject,
-            html,
-            text,
-            forwardedFrom,
-            receivedEmailId,
-          },
+
+        // fallback to forward the email
+        await queueEmailJob(EmailJobType.ForwardEmail, {
+          from,
+          to,
+          subject,
+          html,
+          text,
+          headers,
+          forwardedFrom,
+          replyTo,
         })
+
         res.send('ok')
       }
     } catch (e) {
-      console.log(e)
+      console.error(e)
       res.send(e)
     }
   }
diff --git a/packages/inbound-email-handler/src/job.ts b/packages/inbound-email-handler/src/job.ts
new file mode 100644
index 000000000..89a2fea40
--- /dev/null
+++ b/packages/inbound-email-handler/src/job.ts
@@ -0,0 +1,72 @@
+import { BulkJobOptions, Queue } from 'bullmq'
+import { redisDataSource } from './redis_data_source'
+
+const QUEUE_NAME = 'omnivore-backend-queue'
+export enum EmailJobType {
+  ForwardEmail = 'forward-email',
+  SaveNewsletter = 'save-newsletter',
+  ConfirmationEmail = 'confirmation-email',
+  SaveAttachment = 'save-attachment',
+}
+
+interface EmailJobData {
+  from: string
+  to: string
+  subject: string
+  html?: string
+  text?: string
+  headers?: Record<string, string | string[]>
+  unsubMailTo?: string
+  unsubHttpUrl?: string
+  forwardedFrom?: string
+  replyTo?: string
+  uploadFile?: {
+    fileName: string
+    contentType: string
+    id: string
+  }
+  confirmationCode?: string
+}
+
+const queue = new Queue(QUEUE_NAME, {
+  connection: redisDataSource.queueRedisClient,
+})
+
+const getPriority = (jobType: EmailJobType): number => {
+  // we want to prioritized jobs by the expected time to complete
+  // lower number means higher priority
+  // priority 1: jobs that are expected to finish immediately
+  // priority 5: jobs that are expected to finish in less than 10 second
+  // priority 10: jobs that are expected to finish in less than 10 minutes
+  // priority 100: jobs that are expected to finish in less than 1 hour
+  switch (jobType) {
+    case EmailJobType.ForwardEmail:
+    case EmailJobType.ConfirmationEmail:
+      return 1
+    case EmailJobType.SaveAttachment:
+    case EmailJobType.SaveNewsletter:
+      return 5
+    default:
+      throw new Error(`unknown job type: ${jobType as string}`)
+  }
+}
+
+const getOpts = (jobType: EmailJobType): BulkJobOptions => {
+  return {
+    removeOnComplete: true,
+    removeOnFail: true,
+    attempts: 3,
+    priority: getPriority(jobType),
+    backoff: {
+      type: 'exponential',
+      delay: 2000,
+    },
+  }
+}
+
+export const queueEmailJob = async (
+  jobType: EmailJobType,
+  data: EmailJobData
+) => {
+  await queue.add(jobType, data, getOpts(jobType))
+}
diff --git a/packages/inbound-email-handler/src/newsletter.ts b/packages/inbound-email-handler/src/newsletter.ts
index 54be5d3dd..b31103ada 100644
--- a/packages/inbound-email-handler/src/newsletter.ts
+++ b/packages/inbound-email-handler/src/newsletter.ts
@@ -1,12 +1,11 @@
 import addressparser from 'addressparser'
-import { publishMessage } from './index'
+import { EmailJobType, queueEmailJob } from './job'
 
 interface Unsubscribe {
   mailTo?: string
   httpUrl?: string
 }
 
-const GOOGLE_CONFIRMATION_CODE_RECEIVED_TOPIC = 'emailConfirmationCodeReceived'
 const GOOGLE_CONFIRMATION_EMAIL_SENDER_ADDRESS = 'forwarding-noreply@google.com'
 // check unicode parentheses too
 const GOOGLE_CONFIRMATION_CODE_PATTERN = /\d+/u
@@ -41,24 +40,25 @@ export const parseAuthor = (address: string): string => {
 }
 
 export const handleGoogleConfirmationEmail = async (
-  email: string,
+  from: string,
+  to: string,
   subject: string
 ) => {
-  console.log('confirmation email', email, subject)
+  console.log('confirmation email', from, to, subject)
 
   const confirmationCode = getConfirmationCode(subject)
-  if (!email || !confirmationCode) {
+  if (!to || !confirmationCode) {
     console.log(
       'confirmation email error, user email:',
-      email,
+      to,
       'confirmationCode',
       confirmationCode
     )
     throw new Error('invalid confirmation email')
   }
 
-  const message = { emailAddress: email, confirmationCode: confirmationCode }
-  return publishMessage(GOOGLE_CONFIRMATION_CODE_RECEIVED_TOPIC, message)
+  const message = { from, to, confirmationCode, subject }
+  return queueEmailJob(EmailJobType.ConfirmationEmail, message)
 }
 
 export const getConfirmationCode = (subject: string): string | undefined => {
diff --git a/packages/inbound-email-handler/src/redis_data_source.ts b/packages/inbound-email-handler/src/redis_data_source.ts
new file mode 100644
index 000000000..05bf8d26c
--- /dev/null
+++ b/packages/inbound-email-handler/src/redis_data_source.ts
@@ -0,0 +1,99 @@
+import Redis, { RedisOptions } from 'ioredis'
+import 'dotenv/config'
+
+type RedisClientType = 'cache' | 'mq'
+type RedisDataSourceOption = {
+  url?: string
+  cert?: string
+}
+export type RedisDataSourceOptions = {
+  [key in RedisClientType]: RedisDataSourceOption
+}
+
+export class RedisDataSource {
+  options: RedisDataSourceOptions
+
+  cacheClient: Redis
+  queueRedisClient: Redis
+
+  constructor(options: RedisDataSourceOptions) {
+    this.options = options
+
+    const cacheClient = createIORedisClient('cache', this.options)
+    if (!cacheClient) throw 'Error initializing cache redis client'
+
+    this.cacheClient = cacheClient
+    this.queueRedisClient =
+      createIORedisClient('mq', this.options) || this.cacheClient // if mq is not defined, use cache
+  }
+
+  async shutdown(): Promise<void> {
+    try {
+      await this.queueRedisClient?.quit()
+      await this.cacheClient?.quit()
+    } catch (err) {
+      console.error('error while shutting down redis', err)
+    }
+  }
+}
+
+const createIORedisClient = (
+  name: RedisClientType,
+  options: RedisDataSourceOptions
+): Redis | undefined => {
+  const option = options[name]
+  const redisURL = option.url
+  if (!redisURL) {
+    console.log(`no redisURL supplied: ${name}`)
+    return undefined
+  }
+
+  const redisCert = option.cert
+  const tls =
+    redisURL.startsWith('rediss://') && redisCert
+      ? {
+          ca: redisCert,
+          rejectUnauthorized: false,
+        }
+      : undefined
+
+  const redisOptions: RedisOptions = {
+    tls,
+    name,
+    connectTimeout: 10000,
+    maxRetriesPerRequest: null,
+    offlineQueue: false,
+  }
+  return new Redis(redisURL, redisOptions)
+}
+
+export const redisDataSource = new RedisDataSource({
+  cache: {
+    url: process.env.REDIS_URL,
+    cert: process.env.REDIS_CERT,
+  },
+  mq: {
+    url: process.env.MQ_REDIS_URL,
+    cert: process.env.MQ_REDIS_CERT,
+  },
+})
+
+const gracefulShutdown = async (signal: string) => {
+  console.log(`Received ${signal}, shutting down gracefully...`)
+
+  await redisDataSource.shutdown()
+  console.log('redis shutdown successfully')
+
+  process.exit(0)
+}
+
+process.on('SIGINT', () => {
+  ;(async () => {
+    await gracefulShutdown('SIGINT')
+  })()
+})
+process.on('SIGTERM', () => {
+  ;(async () => {
+    await gracefulShutdown('SIGTERM')
+  })()
+})
diff --git a/packages/inbound-email-handler/test/global-teardown.ts b/packages/inbound-email-handler/test/global-teardown.ts
new file mode 100644
index 000000000..cb733d056
--- /dev/null
+++ b/packages/inbound-email-handler/test/global-teardown.ts
@@ -0,0 +1,5 @@
+import { redisDataSource } from '../src/redis_data_source'
+
+export const mochaGlobalTeardown = async () => {
+  await redisDataSource.shutdown()
+}
diff --git a/packages/inbound-email-handler/test/newsletter.test.ts b/packages/inbound-email-handler/test/newsletter.test.ts
index cf9bff45c..b32a7f504 100644
--- a/packages/inbound-email-handler/test/newsletter.test.ts
+++ b/packages/inbound-email-handler/test/newsletter.test.ts
@@ -2,7 +2,7 @@ import { expect } from 'chai'
 import 'mocha'
 import parseHeaders from 'parse-headers'
 import rfc2047 from 'rfc2047'
-import { parsedTo, plainTextToHtml } from '../src'
+import { parsedTo } from '../src'
 import {
   getConfirmationCode,
   isGoogleConfirmationEmail,
@@ -138,29 +138,3 @@ describe('decode and parse headers', () => {
     })
   })
 })
-
-describe('plainTextToHtml', () => {
-  it('converts text to html', () => {
-    const text =
-      'DEVOPS WEEKLY\r\n' +
-      'ISSUE #665 - 24th September 2023\r\n' +
-      '\r\n' +
-      'A few posts on CI tooling this week, along with a good introduction to developer portals/platforms and other topics.\r\n' +
-      '\r\n' +
-      'StackHawk sponsors Devops Weekly\r\n' +
-      '============================\r\n' +
-      '\r\n' +
-      'Experience automated security testing without the hassle of connecting your own app or configuring an environment! Follow the Tutorial to try out StackHawk and explore a world where security becomes an accelerator, not a blocker\r\n' +
-      '\r\n' +
-      'https://sthwk.com/tutorial\r\n' +
-      '\r\n'
-    expect(plainTextToHtml(text)).to.eql(
-      `<p>DEVOPS WEEKLY
-ISSUE #665 - 24th September 2023</p>
-<p>A few posts on CI tooling this week, along with a good introduction to developer portals/platforms and other topics.</p>
-<h1 id="stackhawksponsorsdevopsweekly">StackHawk sponsors Devops Weekly</h1>
-<p>Experience automated security testing without the hassle of connecting your own app or configuring an environment! Follow the Tutorial to try out StackHawk and explore a world where security becomes an accelerator, not a blocker</p>
-<p>https://sthwk.com/tutorial</p>`
-    )
-  })
-})
diff --git a/packages/inbound-email-handler/tsconfig.json b/packages/inbound-email-handler/tsconfig.json
index 5220d6b3f..f450acf38 100644
--- a/packages/inbound-email-handler/tsconfig.json
+++ b/packages/inbound-email-handler/tsconfig.json
@@ -5,5 +5,5 @@
     "rootDir": ".",
     "lib": ["dom"]
   },
-  "include": ["src"]
+  "include": ["src", "test"]
 }
diff --git a/yarn.lock b/yarn.lock
index 71a8f77bc..b437e5cf1 100644
--- a/yarn.lock
+++ b/yarn.lock
@@ -5729,7 +5729,7 @@
     lodash "^4.17.4"
     read-pkg-up "^7.0.0"
 
-"@sendgrid/client@^7.6.0", "@sendgrid/client@^7.7.0":
+"@sendgrid/client@^7.7.0":
   version "7.7.0"
   resolved "https://registry.yarnpkg.com/@sendgrid/client/-/client-7.7.0.tgz#f8f67abd604205a0d0b1af091b61517ef465fdbf"
   integrity sha512-SxH+y8jeAQSnDavrTD0uGDXYIIkFylCo+eDofVmZLQ0f862nnqbC3Vd1ej6b7Le7lboyzQF6F7Fodv02rYspuA==
@@ -7727,6 +7727,11 @@
   resolved "https://registry.yarnpkg.com/@types/chai/-/chai-4.3.4.tgz#e913e8175db8307d78b4e8fa690408ba6b65dee4"
   integrity sha512-KnRanxnpfpjUTqTCXslZSEdLfXExwgNxYPdiO2WGUj8+HDjFi8R3k5RVKPeSCzLjCcshCAtVO2QBbVuAV4kTnw==
 
+"@types/chai@^4.3.6":
+  version "4.3.14"
+  resolved "https://registry.yarnpkg.com/@types/chai/-/chai-4.3.14.tgz#ae3055ea2be43c91c9fd700a36d67820026d96e6"
+  integrity sha512-Wj71sXE4Q4AkGdG9Tvq1u/fquNz9EdG4LIJMwVVII7ashjD/8cf8fyIfJAjRr6YcsXnSE8cOGQPq1gqeR8z+3w==
+
 "@types/chrome@^0.0.197":
   version "0.0.197"
   resolved "https://registry.yarnpkg.com/@types/chrome/-/chrome-0.0.197.tgz#c1b50cdb72ee40f9bc1411506031a9f8a925ab35"
@@ -8258,6 +8263,11 @@
   resolved "https://registry.yarnpkg.com/@types/minimist/-/minimist-1.2.5.tgz#ec10755e871497bcd83efe927e43ec46e8c0747e"
   integrity sha512-hov8bUuiLiyFPGyFPE1lwWhmzYbirOXQNNo40+y3zow8aFVTeyn3VWL0VFFfdNddA8S4Vf0Tc062rzyNr7Paag==
 
+"@types/mocha@^10.0.0":
+  version "10.0.6"
+  resolved "https://registry.yarnpkg.com/@types/mocha/-/mocha-10.0.6.tgz#818551d39113081048bdddbef96701b4e8bb9d1b"
+  integrity sha512-dJvrYWxP/UcXm36Qn36fxhUKu8A/xMRXVT2cliFF1Z7UA9liG5Psj3ezNSZw+5puH2czDXRLcXQxf8JbJt0ejg==
+
 "@types/mocha@^10.0.1":
   version "10.0.1"
   resolved "https://registry.yarnpkg.com/@types/mocha/-/mocha-10.0.1.tgz#2f4f65bb08bc368ac39c96da7b2f09140b26851b"