import { getGmailAccessToken } from "../gmail"
import { logger } from "../logger"
import { sendGmailReply } from "./emailService"
import { prisma } from "../prisma"
import { renderTemplate } from "@/lib/templates/simpleRenderer"
import { classifyCustomerEmail } from "@/lib/ai/gmailGroq"
import { screenInboundEmailForAbuse } from "@/lib/ai/gmailSafety"
import { buildOrderSummaryForEmail, extractOrderNumberHints } from "@/lib/ai/gmailContext"
import { findTemplateWithFallback, type TemplateRecord } from "@/lib/juno/templatesRepo"
export type ProcessIncomingEmailResult = {
success: boolean
error?: string
/** True when the pipeline attempted an auto-reply to a qualifying unread customer message */
replied?: boolean
/** True when the message was recorded as abusive/suspicious and no auto-reply was sent */
flagged?: boolean
}
/** Normalize Gmail / RFC5322 From header to a lowercase email address. */
function parseSenderAddress(fromHeader: string): string {
const raw = fromHeader.trim()
const inAngles = raw.match(/<([^>]+)>/)
if (inAngles) {
return inAngles[1].trim().replace(/^mailto:/i, "").toLowerCase()
}
const lone = raw.replace(/^mailto:/i, "").trim()
if (/^[^\s<>"']+@[^\s<>"']+$/.test(lone)) {
return lone.toLowerCase()
}
const tokens = raw.split(/\s+/)
for (let i = tokens.length - 1; i >= 0; i--) {
const t = tokens[i].replace(/[<>",]/g, "").trim()
if (t.includes("@") && !t.includes("..")) {
return t.replace(/^mailto:/i, "").toLowerCase()
}
}
return lone.toLowerCase()
}
/** Local-part of email, title-cased for greetings (templates use {{customer_name}}). */
function greetingNameFromEmail(email: string): string {
const local = email.split("@")[0]?.trim() || ""
if (!local) return "there"
return local.charAt(0).toUpperCase() + local.slice(1).toLowerCase()
}
function buildMinimalFallbackHtml(emailBody: string, storeName: string): string {
const safeSnippet = emailBody.replace(/<[^>]+>/g, "").slice(0, 400)
return `
Thanks for reaching out
We have received your message and our team will review it shortly.
Summary of what you sent:
${safeSnippet || "No additional message content was detected."}
You can reply to this email if you have more details to share.
Best regards,
${storeName} Support
`
}
/** When no template exists but we need an order # before sharing order rows. */
function buildNeedOrderIdFallbackHtml(storeName: string, storeEmail: string): string {
const contact = storeEmail.trim()
? `Reply with your order number, or email ${storeEmail}.`
: `Reply with your order number from your confirmation email.`
return `
Hello,
We didn't see an order number in your message. To look up the right order for ${storeName},
please reply with your order number (for example #1002 or 1002) from your order confirmation.
${contact}
Best regards,
${storeName} Support
`
}
/** When no off_topic template exists — polite boundary, no order data. */
function buildOffTopicFallbackHtml(storeName: string, storeEmail: string): string {
const contact = storeEmail.trim()
? `If you have a question about an order from ${storeName}, reply with details or email us at ${storeEmail}.`
: `If you have a question about an order from ${storeName}, reply with your question and order number if you have one.`
return `
Hello,
This inbox only handles orders, products, shipping, returns, and billing for ${storeName}.
We can't answer unrelated requests (news, weather, other websites, or general topics).
${contact}
Best regards,
${storeName} Support
`
}
function renderFromTemplate(
template: TemplateRecord,
context: Record
): string {
return template.isHtml
? renderTemplate(template.body, context)
: `
${renderTemplate(
template.body,
context,
)}
`
}
/**
* Process an incoming email and generate an AI-assisted reply (Groq classification + templates).
*/
export async function processIncomingEmail(
userId: string,
storeId: string | null,
_historyId?: string,
options?: { replyTrigger?: string }
): Promise {
let replied = false
try {
const accessToken = await getGmailAccessToken(userId)
if (!accessToken) {
logger.warn("Cannot process email - Gmail not connected", { userId })
return { success: false, error: "Gmail not connected", replied: false }
}
if (!storeId) {
logger.info("No active store found for user, skipping email processing", { userId })
return { success: false, error: "No active store", replied: false }
}
const unreadEmails = await getUnreadEmails(accessToken, 1)
if (unreadEmails.length === 0) {
logger.info("No unread emails found", { userId })
return { success: true, replied: false }
}
const email = unreadEmails[0]
const senderEmail = parseSenderAddress(email.from)
if (!senderEmail.endsWith("@gmail.com")) {
logger.info("Skipping non-Gmail email", { userId, senderEmail })
return { success: true, replied: false }
}
const fullEmail = await getEmailContent(accessToken, email.id)
if (!fullEmail) {
logger.error("Failed to fetch email content", { emailId: email.id })
return { success: false, error: "Failed to fetch email content", replied: false }
}
const emailBody = extractEmailBody(fullEmail)
const emailSubject = email.subject || "No Subject"
logger.info("Processing incoming email", {
userId,
storeId,
from: senderEmail,
subject: emailSubject,
emailId: email.id,
})
const plainForAi = emailBody.replace(/<[^>]+>/g, " ").replace(/\s+/g, " ").trim()
const safety = await screenInboundEmailForAbuse(emailSubject, plainForAi)
if (safety?.flagged) {
try {
const existing = await prisma.flaggedEmail.findFirst({
where: { storeId, messageId: email.id },
})
if (!existing) {
await prisma.flaggedEmail.create({
data: {
storeId,
subject: emailSubject.slice(0, 5000),
from: senderEmail,
reason: safety.reason,
priority: safety.priority,
messageId: email.id,
},
})
}
logger.warn("Inbound email flagged — no auto-reply", {
userId,
storeId,
from: senderEmail,
priority: safety.priority,
})
} catch (flagErr) {
logger.error("Failed to persist flagged email", flagErr)
}
const markedRead = await markEmailAsRead(accessToken, email.id)
if (!markedRead) {
logger.warn("Failed to mark flagged email as read", { userId, emailId: email.id })
}
return { success: true, replied: false, flagged: true }
}
try {
const incomingPreview = emailBody
.replace(/<[^>]+>/g, "")
.replace(/\s+\n/g, "\n")
.replace(/\n{3,}/g, "\n\n")
.trim()
// eslint-disable-next-line @typescript-eslint/no-explicit-any
await (prisma as any).emailLog.create({
data: {
userId,
storeId,
to: senderEmail,
subject: emailSubject,
bodyPreview: incomingPreview,
isAutomated: false,
trigger: "GMAIL_INCOMING",
status: "RECEIVED",
providerMessageId: email.messageId,
errorMessage: null,
sentAt: new Date(),
},
})
} catch (logError) {
logger.error("Failed to log incoming email to EmailLog", logError)
}
const classification = await classifyCustomerEmail(emailSubject, plainForAi)
const hintSource = `${emailSubject}\n${plainForAi}`
const isOffTopic = classification.intent === "off_topic"
const orderHints = isOffTopic ? [] : extractOrderNumberHints(hintSource)
/** Don’t load CachedOrder until the customer names an order (avoids guessing #1002 when they only say “where’s my package?”). */
const intentsNeedingOrderIdWhenAmbiguous = new Set([
"shipping_delay",
"refund_request",
"account_billing",
])
const needsOrderNumberFirst =
!isOffTopic &&
intentsNeedingOrderIdWhenAmbiguous.has(classification.intent) &&
orderHints.length === 0
let orderSummary = ""
if (!isOffTopic && !needsOrderNumberFirst) {
orderSummary = await buildOrderSummaryForEmail(storeId, senderEmail, hintSource)
}
logger.info("Gmail classification", {
userId,
storeId,
intent: classification.intent,
mood: classification.mood,
confidence: classification.confidence,
})
logger.info("Gmail order context for reply", {
storeId,
orderSummaryChars: orderSummary.trim().length,
orderHints,
offTopic: isOffTopic,
needsOrderNumberFirst,
})
const store = await prisma.store.findUnique({
where: { id: storeId },
select: {
businessName: true,
email: true,
},
})
const storeName = store?.businessName || "Your store"
const ticketSummary =
classification.short_summary?.trim() ||
plainForAi.slice(0, 500) ||
emailBody.slice(0, 500)
const offTopicOrderLine = `We don't include order or purchase history for unrelated topics. For help with ${storeName}, reply with a store-related question and your order number if you have one.`
const counterOrderIdLine = `No order details are shown yet — we did not see an order number in your message. Please reply with your order number (e.g. #1002 or 1002) from your ${storeName} confirmation email so we can look up the correct order.`
const context: Record = {
customer_name: greetingNameFromEmail(senderEmail),
customer_email: senderEmail,
subject: emailSubject,
store_name: storeName,
store_email: store?.email || "",
ticket_summary: ticketSummary,
order_summary: isOffTopic
? offTopicOrderLine
: needsOrderNumberFirst
? counterOrderIdLine
: orderSummary || "No matching recent orders on file for this email.",
}
let finalReply: string | null = null
try {
const template = await findTemplateWithFallback(
storeId,
classification.intent,
classification.mood,
)
if (template) {
finalReply = renderFromTemplate(template, context)
}
} catch (replyBuildError) {
logger.error("Failed to load/render template for auto-reply", replyBuildError)
}
if (!finalReply) {
if (isOffTopic) {
finalReply = buildOffTopicFallbackHtml(storeName, store?.email || "")
} else if (needsOrderNumberFirst) {
finalReply = buildNeedOrderIdFallbackHtml(storeName, store?.email || "")
} else {
finalReply = buildMinimalFallbackHtml(emailBody, storeName)
}
}
const replySubject = emailSubject.startsWith("Re:") ? emailSubject : `Re: ${emailSubject}`
replied = true
const result = await sendGmailReply(
userId,
senderEmail,
replySubject,
finalReply,
email.threadId,
{
storeId,
isAutomated: true,
trigger: options?.replyTrigger ?? "GMAIL_WEBHOOK",
originalMessageId: email.messageId,
}
)
const markedRead = await markEmailAsRead(accessToken, email.id)
if (!markedRead) {
logger.warn("Failed to mark email as read after processing", {
userId,
storeId,
emailId: email.id,
})
}
if (result.success) {
logger.info("Automated reply sent successfully", {
userId,
storeId,
to: senderEmail,
subject: emailSubject,
})
} else {
logger.error("Failed to send automated reply", {
userId,
storeId,
to: senderEmail,
subject: emailSubject,
error: result.error,
})
}
return { success: result.success, error: result.error, replied }
} catch (error) {
logger.error("Error processing incoming email", error)
return {
success: false,
error: error instanceof Error ? error.message : "Unknown error",
replied,
}
}
}
/**
* Get unread emails from Gmail
*/
async function getUnreadEmails(
accessToken: string,
maxResults: number = 10
): Promise<
Array<{
id: string
threadId: string
from: string
subject: string
snippet: string
messageId: string
}>
> {
try {
const searchResponse = await fetch(
`https://gmail.googleapis.com/gmail/v1/users/me/messages?q=is:unread newer_than:10m&maxResults=${maxResults}`,
{
headers: {
Authorization: `Bearer ${accessToken}`,
},
}
)
if (!searchResponse.ok) {
logger.error("Failed to search Gmail messages")
return []
}
const searchResult = await searchResponse.json()
const messages = searchResult.messages || []
const emailDetails = await Promise.all(
messages.map(async (msg: { id: string; threadId: string }) => {
try {
const messageResponse = await fetch(
`https://gmail.googleapis.com/gmail/v1/users/me/messages/${msg.id}?format=metadata&metadataHeaders=From&metadataHeaders=Subject&metadataHeaders=Date&metadataHeaders=Message-ID`,
{
headers: {
Authorization: `Bearer ${accessToken}`,
},
}
)
if (!messageResponse.ok) {
return null
}
const messageData = await messageResponse.json()
const headers = messageData.payload?.headers || []
const fromHeader = headers.find((h: { name: string }) => h.name.toLowerCase() === "from")
const subjectHeader = headers.find((h: { name: string }) => h.name.toLowerCase() === "subject")
const messageIdHeader = headers.find(
(h: { name: string }) => h.name.toLowerCase() === "message-id"
)
return {
id: msg.id,
threadId: msg.threadId,
from: fromHeader?.value || "",
subject: subjectHeader?.value || "",
snippet: messageData.snippet || "",
messageId: messageIdHeader?.value || "",
}
} catch (error) {
logger.error("Error fetching message details", { error, messageId: msg.id })
return null
}
})
)
return emailDetails.filter((e) => e !== null) as Array<{
id: string
threadId: string
from: string
subject: string
snippet: string
messageId: string
}>
} catch (error) {
logger.error("Error getting unread emails", error)
return []
}
}
async function getEmailContent(accessToken: string, messageId: string): Promise {
try {
const response = await fetch(
`https://gmail.googleapis.com/gmail/v1/users/me/messages/${messageId}?format=full`,
{
headers: {
Authorization: `Bearer ${accessToken}`,
},
}
)
if (!response.ok) {
return null
}
return await response.json()
} catch (error) {
logger.error("Error fetching email content", { error, messageId })
return null
}
}
function extractEmailBody(message: {
payload?: unknown
snippet?: string
}): string {
try {
const payload = message.payload as
| {
mimeType?: string
body?: { data?: string }
parts?: unknown[]
}
| undefined
if (!payload) {
return ""
}
const decodePart = (part: { body?: { data?: string } }): string => {
if (part.body?.data) {
return Buffer.from(part.body.data, "base64url").toString("utf-8")
}
return ""
}
const collectParts = (part: Record, mimeType: string): string[] => {
const results: string[] = []
if (part.mimeType === mimeType && (part.body as { data?: string } | undefined)?.data) {
results.push(decodePart(part as { body?: { data?: string } }))
}
if (Array.isArray(part.parts)) {
for (const p of part.parts) {
results.push(...collectParts(p as Record, mimeType))
}
}
return results
}
let body =
collectParts(payload as Record, "text/plain").join("\n").trim() ||
collectParts(payload as Record, "text/html").join("\n").trim() ||
message.snippet ||
""
const lines = body.split(/\r?\n/)
const cleanedLines: string[] = []
for (const line of lines) {
const trimmed = line.trim()
if (trimmed.startsWith(">")) continue
if (/^on .+wrote:$/i.test(trimmed)) break
cleanedLines.push(line)
}
body = cleanedLines.join("\n").trim()
return body
} catch (error) {
logger.error("Error extracting email body", error)
return ""
}
}
async function markEmailAsRead(accessToken: string, messageId: string): Promise {
try {
const response = await fetch(
`https://gmail.googleapis.com/gmail/v1/users/me/messages/${messageId}/modify`,
{
method: "POST",
headers: {
Authorization: `Bearer ${accessToken}`,
"Content-Type": "application/json",
},
body: JSON.stringify({
removeLabelIds: ["UNREAD"],
}),
}
)
return response.ok
} catch (error) {
logger.error("Error marking email as read", error)
return false
}
}