Run email triggers in jobs (#1032)

Author: csansoon

packages/core/src/jobs/constants.ts

@@ -34,6 +34,7 @@ export enum Jobs {
   autoScaleJob = 'autoScaleJob',
   updateMcpServerLastUsedJob = 'updateMcpServerLastUsedJob',
   scaleDownMcpServerJob = 'scaleDownMcpServerJob',
+  runEmailTriggerJob = 'runEmailTriggerJob',
 }
 
 export const QUEUES = {
@@ -53,6 +54,7 @@ export const QUEUES = {
       'requestDocumentSuggestionsJob',
       'checkScheduledDocumentTriggersJob',
       'processScheduledTriggerJob',
+      'runEmailTriggerJob',
     ],
   },
   [Queues.maintenanceQueue]: {

packages/core/src/jobs/job-definitions/documentTriggers/index.ts

@@ -1,4 +1,9 @@
 import { checkScheduledDocumentTriggersJob } from './checkScheduledDocumentTriggersJob'
 import { processScheduledTriggerJob } from './processScheduledTriggerJob'
+import { runEmailTriggerJob } from './runEmailTriggerJob'
 
-export { checkScheduledDocumentTriggersJob, processScheduledTriggerJob }
+export {
+  checkScheduledDocumentTriggersJob,
+  processScheduledTriggerJob,
+  runEmailTriggerJob,
+}

packages/core/src/services/documentTriggers/handlers/email/getResponse.ts renamed to packages/core/src/jobs/job-definitions/documentTriggers/runEmailTriggerJob/getResponse.ts

@@ -16,7 +16,7 @@ import {
   DocumentVersionsRepository,
 } from '../../../../repositories'
 import { database } from '../../../../client'
-import { runDocumentAtCommit } from '../../../commits/runDocumentAtCommit'
+import { runDocumentAtCommit } from '../../../../services/commits'
 import { unsafelyFindWorkspace } from '../../../../data-access'
 import { DocumentTrigger, Workspace } from '../../../../browser'
 import {
@@ -25,9 +25,9 @@ import {
   type AssistantMessage,
   type UserMessage,
 } from '@latitude-data/compiler'
-import { addMessages } from '../../../documentLogs'
-import { uploadFile } from '../../../files'
-import { EmailTriggerConfiguration } from '../../helpers/schema'
+import { addMessages } from '../../../../services/documentLogs'
+import { uploadFile } from '../../../../services/files'
+import { EmailTriggerConfiguration } from '../../../../services/documentTriggers/helpers/schema'
 
 async function getNewTriggerResponse(
   {
@@ -216,7 +216,7 @@ export async function getEmailResponse(
     senderName,
     subject,
     body,
-    attachments: attachedFiles,
+    attachments,
   }: {
     documentUuid: string
     trigger: DocumentTrigger
@@ -226,7 +226,7 @@ export async function getEmailResponse(
     senderName: string | undefined
     subject: string
     body: string
-    attachments?: File[]
+    attachments?: PromptLFile[]
   },
   db = database,
 ): PromisedResult<AssistantMessage, LatitudeError> {
@@ -235,13 +235,6 @@ export async function getEmailResponse(
     db,
   )) as Workspace
 
-  const attachmentsResult = await uploadAttachments({
-    workspace,
-    attachments: attachedFiles ?? [],
-  })
-  if (attachmentsResult.error) return attachmentsResult
-  const attachments = attachmentsResult.unwrap()
-
   const referencedLogResult = await findReferencedLog(
     {
       workspace,

packages/core/src/jobs/job-definitions/documentTriggers/runEmailTriggerJob/index.ts

@@ -0,0 +1,120 @@
+import { Job } from 'bullmq'
+import { EMAIL_TRIGGER_DOMAIN } from '@latitude-data/constants'
+import { PromptLFile } from 'promptl-ai'
+import { unsafelyFindWorkspace } from '../../../../data-access'
+import { DocumentTrigger, HEAD_COMMIT, Workspace } from '../../../../browser'
+import {
+  DocumentTriggersRepository,
+  DocumentVersionsRepository,
+} from '../../../../repositories'
+import { getEmailResponse } from './getResponse'
+import { EmailTriggerConfiguration } from '../../../../services/documentTriggers/helpers/schema'
+import { LatitudeError, PromisedResult, Result } from '../../../../lib'
+import { DocumentTriggerMailer } from '../../../../mailers'
+
+export type RunEmailTriggerJobData = {
+  workspaceId: number
+  triggerId: number
+  recipient: string
+  senderEmail: string
+  senderName: string | undefined
+  messageId?: string
+  parentMessageIds?: string[]
+  subject: string
+  body: string
+  attachments: PromptLFile[]
+}
+
+async function getTriggerName(
+  trigger: DocumentTrigger,
+): PromisedResult<string, LatitudeError> {
+  const configName = (
+    trigger.configuration as EmailTriggerConfiguration
+  ).name?.trim()
+  if (configName?.length) {
+    return Result.ok(configName)
+  }
+
+  const docsScope = new DocumentVersionsRepository(trigger.workspaceId)
+  const documentResult = await docsScope.getDocumentAtCommit({
+    projectId: trigger.projectId,
+    commitUuid: HEAD_COMMIT,
+    documentUuid: trigger.documentUuid,
+  })
+  if (documentResult.error) return documentResult
+
+  const document = documentResult.unwrap()
+  const docName = document.path.split('/').pop()!
+  return Result.ok(docName)
+}
+
+export const runEmailTriggerJob = async (job: Job<RunEmailTriggerJobData>) => {
+  const {
+    workspaceId,
+    triggerId,
+    senderEmail,
+    senderName,
+    messageId,
+    parentMessageIds,
+    subject,
+    body,
+    attachments,
+  } = job.data
+
+  const workspace = (await unsafelyFindWorkspace(workspaceId)) as Workspace
+
+  const triggerScope = new DocumentTriggersRepository(workspace.id)
+  const triggerResult = await triggerScope.find(triggerId)
+  const trigger = triggerResult.unwrap()
+
+  const nameResult = await getTriggerName(trigger)
+  const name = nameResult.unwrap()
+
+  const responseResult = await getEmailResponse({
+    documentUuid: trigger.documentUuid,
+    trigger,
+    messageId,
+    parentMessageIds,
+    senderEmail,
+    senderName,
+    subject,
+    body,
+    attachments,
+  })
+
+  const configuration = trigger.configuration as EmailTriggerConfiguration
+  if (configuration.replyWithResponse === false) {
+    return
+  }
+
+  const from = `${JSON.stringify(name)} <${trigger.documentUuid}@${EMAIL_TRIGGER_DOMAIN}>`
+
+  const references = [
+    ...(parentMessageIds ?? []),
+    ...(messageId ? [messageId] : []),
+  ].join(' ')
+
+  const headers = messageId
+    ? {
+        'In-Reply-To': messageId,
+        References: references,
+        // It seems like nodemailer-mailgun-transport is ignoring the "References" header.
+        // Maybe some of these will work:
+        'X-Mailgun-References': references,
+        'h:References': references,
+        'h:X-Mailgun-References': references,
+      }
+    : undefined
+
+  const mailer = new DocumentTriggerMailer(responseResult, {
+    to: senderEmail,
+    from,
+    inReplyTo: messageId,
+    references,
+    subject: 'Re: ' + subject,
+    headers,
+  })
+
+  const result = await mailer.send()
+  result.unwrap()
+}

packages/core/src/services/documentTriggers/handlers/email/index.ts

@@ -10,11 +10,14 @@ import {
   EMAIL_TRIGGER_DOMAIN,
 } from '@latitude-data/constants'
 import { database } from '../../../../client'
-import { DocumentTrigger, HEAD_COMMIT } from '../../../../browser'
-import { DocumentTriggerMailer } from '../../../../mailers'
-import { getEmailResponse } from './getResponse'
+import { DocumentTrigger, HEAD_COMMIT, Workspace } from '../../../../browser'
 import { DocumentVersionsRepository } from '../../../../repositories'
 import { EmailTriggerConfiguration } from '../../helpers/schema'
+import { PromptLFile } from 'promptl-ai'
+import { uploadFile } from '../../../files'
+import { unsafelyFindWorkspace } from '../../../../data-access'
+import { setupQueues } from '../../../../jobs'
+import { RunEmailTriggerJobData } from '../../../../jobs/job-definitions/documentTriggers/runEmailTriggerJob'
 
 async function getTriggerName(
   trigger: DocumentTrigger,
@@ -67,6 +70,27 @@ export async function assertTriggerFilters({
   return Result.error(new BadRequestError('Sender is not in whitelist'))
 }
 
+export async function uploadAttachments({
+  workspace,
+  attachments,
+}: {
+  workspace: Workspace
+  attachments: File[]
+}): PromisedResult<PromptLFile[], LatitudeError> {
+  const results = await Promise.all(
+    attachments.map(async (file) => {
+      return await uploadFile({ file, workspace })
+    }),
+  )
+
+  const errors = results.filter((result) => result.error)
+  if (errors.length) {
+    return Result.error(errors[0]!.error! as LatitudeError)
+  }
+
+  return Result.ok(results.map((result) => result.unwrap()))
+}
+
 export async function handleEmailTrigger(
   {
     recipient,
@@ -110,56 +134,41 @@ export async function handleEmailTrigger(
   })
   if (assertFilterResult.error) return assertFilterResult
 
-  const name = await getTriggerName(trigger, db)
-  if (name.error) return name
-
-  const responseResult = await getEmailResponse(
-    {
-      documentUuid: documentUuid!,
-      trigger,
-      messageId,
-      parentMessageIds,
-      senderEmail,
-      senderName,
-      subject,
-      body,
-      attachments,
-    },
+  const workspace = (await unsafelyFindWorkspace(
+    trigger.workspaceId,
     db,
-  )
+  )) as Workspace
 
-  const configuration = trigger.configuration as EmailTriggerConfiguration
-  if (configuration.replyWithResponse === false) return Result.nil()
-
-  const from = `${JSON.stringify(name.unwrap())} <${trigger.documentUuid}@${EMAIL_TRIGGER_DOMAIN}>`
-
-  const references = [
-    ...(parentMessageIds ?? []),
-    ...(messageId ? [messageId] : []),
-  ].join(' ')
-
-  const headers = messageId
-    ? {
-        'In-Reply-To': messageId,
-        References: references,
-        // It seems like nodemailer-mailgun-transport is ignoring the "References" header.
-        // Maybe some of these will work:
-        'X-Mailgun-References': references,
-        'h:References': references,
-        'h:X-Mailgun-References': references,
-      }
-    : undefined
-
-  const mailer = new DocumentTriggerMailer(responseResult, {
-    to: senderEmail,
-    from,
-    inReplyTo: messageId,
-    references,
-    subject: 'Re: ' + subject,
-    headers,
+  const nameResult = await getTriggerName(trigger, db)
+  if (nameResult.error) return nameResult
+
+  const uploadResult = await uploadAttachments({
+    workspace,
+    attachments: attachments ?? [],
   })
+  if (uploadResult.error) return uploadResult
+  const uploadedFiles = uploadResult.unwrap()
+
+  const jobQueues = await setupQueues()
+
+  const runJobData: RunEmailTriggerJobData = {
+    workspaceId: workspace.id,
+    triggerId: trigger.id,
+    recipient: recipient,
+    senderEmail: senderEmail,
+    senderName: senderName,
+    messageId: messageId,
+    parentMessageIds: parentMessageIds,
+    subject: subject,
+    body: body,
+    attachments: uploadedFiles,
+  }
+
+  const job =
+    await jobQueues.defaultQueue.jobs.enqueueRunEmailTriggerJob(runJobData)
+  if (!job.id) {
+    return Result.error(new LatitudeError('Failed to enqueue job'))
+  }
 
-  const sendResult = await mailer.send()
-  if (sendResult.error) return sendResult
   return Result.nil()
 }