diff --git a/apps/queue/src/inbound-email/processor.ts b/apps/queue/src/inbound-email/processor.ts new file mode 100644 index 000000000..cfef3d43e --- /dev/null +++ b/apps/queue/src/inbound-email/processor.ts @@ -0,0 +1,431 @@ +import mongoose from "mongoose"; +import crypto from "crypto"; +import { verifyReplyToken } from "@courselit/common-logic"; +import { + CommunityCommentSchema, + ProductDiscussionCommentSchema, + ProductDiscussionReplySchema, +} from "@courselit/orm-models"; +import { stripQuotedContent } from "./provider"; +import type { InboundEmailPayload } from "./provider"; + +/** + * Represents a parsed reply from an inbound email. + */ +interface ParsedReply { + /** The verified reply token payload */ + token: { + userId: string; + domainId: string; + entityId: string; + entityType: "community" | "product"; + commentId?: string; + parentReplyId?: string; + }; + /** The stripped text content to save as the reply */ + content: string; + /** The sender email address */ + senderEmail: string; +} + +/** + * Result of processing an inbound reply email. + */ +interface ProcessInboundReplyResult { + success: boolean; + error?: string; + replyContent?: string; +} + +/** + * Extract the reply token from the To address. + * Format: reply+@ + */ +function extractTokenFromToAddress(to: string): string | null { + if (!to) return null; + + // Match reply+@domain + const match = to.match(/^reply\+([A-Za-z0-9\-_:]+)@/); + return match ? match[1] : null; +} + +/** + * Process an inbound reply email. + * + * Steps: + * 1. Extract reply token from To address + * 2. Verify and decrypt the token + * 3. Validate sender email matches token user + * 4. Find the target discussion/comment + * 5. Strip quoted content from the email body + * 6. Create the reply/comment in the correct collection + */ +export async function processInboundReply( + email: InboundEmailPayload, +): Promise { + try { + // Step 1: Extract token + const tokenStr = extractTokenFromToAddress(email.to); + if (!tokenStr) { + return { + success: false, + error: "No reply token found in To address", + }; + } + + // Step 2: Verify token + const token = verifyReplyToken(tokenStr); + if (!token) { + return { + success: false, + error: "Invalid or expired reply token", + }; + } + + // Step 3: Validate sender email + const senderEmail = email.from.toLowerCase().trim(); + + // Look up the user associated with this token + const UserModel = (mongoose.models.User || + mongoose.model("User", new mongoose.Schema({}, { strict: false }))) as mongoose.Model; + + const user = await UserModel.findOne({ + domain: new mongoose.Types.ObjectId(token.domainId), + userId: token.userId, + }).lean(); + + if (!user) { + return { + success: false, + error: "User not found for this reply token", + }; + } + + // Compare sender email with user's email + const userEmail = (user.email || "").toLowerCase().trim(); + if (senderEmail !== userEmail) { + return { + success: false, + error: + "Sender email does not match the user associated with this reply token", + }; + } + + // Step 4: Strip quoted content + const replyContent = stripQuotedContent(email.text); + if (!replyContent) { + return { + success: false, + error: "No reply content found after stripping quoted text", + }; + } + + // Step 5: Create the reply in the appropriate collection + // Validate: target discussion/comment exists + if (token.entityType === "community") { + const exists = await validateCommunityTarget(token); + if (!exists) { + return { + success: false, + error: "Target discussion post or comment not found", + }; + } + await createCommunityReply(token, replyContent, user.userId); + } else if (token.entityType === "product") { + const exists = await validateProductTarget(token); + if (!exists) { + return { + success: false, + error: "Target product discussion not found", + }; + } + await createProductDiscussionReply(token, replyContent, user.userId); + } + + return { + success: true, + replyContent, + }; + } catch (err: any) { + return { + success: false, + error: `Failed to process reply: ${err.message}`, + }; + } +} + +async function validateCommunityTarget( + token: ParsedReply["token"], +): Promise { + const CommunityPostModel = + (mongoose.models.CommunityPost || + mongoose.model( + "CommunityPost", + new mongoose.Schema({}, { strict: false }), + )) as mongoose.Model; + + const post = await CommunityPostModel.findOne({ + domain: new mongoose.Types.ObjectId(token.domainId), + postId: token.entityId, + }).lean(); + + if (!post) return false; + + if (token.commentId) { + const CommunityCommentModel = + (mongoose.models.CommunityComment || + mongoose.model("CommunityComment", CommunityCommentSchema)) as mongoose.Model; + + const comment = await CommunityCommentModel.findOne({ + domain: new mongoose.Types.ObjectId(token.domainId), + commentId: token.commentId, + }).lean(); + + if (!comment) return false; + } + + return true; +} + +async function validateProductTarget( + token: ParsedReply["token"], +): Promise { + const CourseModel = + (mongoose.models.Course || + mongoose.model("Course", new mongoose.Schema({}, { strict: false }))) as mongoose.Model; + + const course = await CourseModel.findOne({ + domain: new mongoose.Types.ObjectId(token.domainId), + courseId: token.entityId, + }).lean(); + + if (!course) return false; + + if (token.commentId) { + const ProductDiscussionCommentModel = + (mongoose.models.ProductDiscussionComment || + mongoose.model( + "ProductDiscussionComment", + ProductDiscussionCommentSchema, + )) as mongoose.Model; + + const comment = await ProductDiscussionCommentModel.findOne({ + domain: new mongoose.Types.ObjectId(token.domainId), + commentId: token.commentId, + }).lean(); + + if (!comment) return false; + } + + return true; +} + +/** + * Creates a reply in a Community discussion. + */ +async function createCommunityReply( + token: ParsedReply["token"], + content: string, + userId: string, +): Promise { + const CommunityCommentModel = + (mongoose.models.CommunityComment || + mongoose.model("CommunityComment", CommunityCommentSchema)) as mongoose.Model; + + if (token.commentId) { + // Replying to an existing comment (as a reply within the comment) + const replyObj = { + userId, + content, + replyId: crypto.randomUUID(), + parentReplyId: token.parentReplyId || null, + media: [], + likes: [], + deleted: false, + }; + + await CommunityCommentModel.updateOne( + { + domain: new mongoose.Types.ObjectId(token.domainId), + commentId: token.commentId, + }, + { + $push: { replies: replyObj }, + }, + ); + + // Dispatch notification for the new reply + await dispatchReplyNotification({ + domain: token.domainId, + userId, + activityType: "community_reply_created", + entityId: replyObj.replyId, + entityTargetId: token.commentId, + metadata: { + commentId: token.commentId, + postId: token.entityId, + }, + }); + } else { + // Creating a top-level comment on the post + const CommunityPostModel = + (mongoose.models.CommunityPost || + mongoose.model( + "CommunityPost", + new mongoose.Schema({}, { strict: false }), + )) as mongoose.Model; + const post = await CommunityPostModel.findOne({ + domain: new mongoose.Types.ObjectId(token.domainId), + postId: token.entityId, + }).lean(); + + const commentObj = { + domain: new mongoose.Types.ObjectId(token.domainId), + userId, + communityId: post?.communityId || "", + postId: token.entityId, + commentId: crypto.randomUUID(), + content, + media: [], + likes: [], + replies: [], + deleted: false, + }; + + await CommunityCommentModel.create(commentObj); + + await dispatchReplyNotification({ + domain: token.domainId, + userId, + activityType: "community_comment_created", + entityId: commentObj.commentId, + metadata: { + postId: token.entityId, + }, + }); + } +} + +/** + * Creates a reply in a Product (Course) discussion. + */ +async function createProductDiscussionReply( + token: ParsedReply["token"], + content: string, + userId: string, +): Promise { + if (!token.commentId) { + // No commentId means this is a top-level comment on a product lesson + const ProductDiscussionCommentModel = + (mongoose.models.ProductDiscussionComment || + mongoose.model( + "ProductDiscussionComment", + ProductDiscussionCommentSchema, + )) as mongoose.Model; + + const commentObj = { + domain: new mongoose.Types.ObjectId(token.domainId), + productId: token.entityId, + entityType: "lesson", + entityId: token.entityId, + commentId: crypto.randomUUID(), + userId, + content: [{ type: "paragraph", children: [{ text: content }] }], + likesCount: 0, + deleted: false, + isEdited: false, + }; + + await ProductDiscussionCommentModel.create(commentObj); + + await dispatchReplyNotification({ + domain: token.domainId, + userId, + activityType: "course_discussion_comment_created", + entityId: commentObj.commentId, + metadata: { + courseId: token.entityId, + entityType: "lesson", + entityId: token.entityId, + eventType: "comment_created", + contentType: "comment", + }, + }); + return; + } + + // Replying to an existing comment + const ProductDiscussionReplyModel = + (mongoose.models.ProductDiscussionReply || + mongoose.model("ProductDiscussionReply", ProductDiscussionReplySchema)) as mongoose.Model; + + const ProductDiscussionCommentModel = + (mongoose.models.ProductDiscussionComment || + mongoose.model( + "ProductDiscussionComment", + ProductDiscussionCommentSchema, + )) as mongoose.Model; + const parentComment = await ProductDiscussionCommentModel.findOne({ + domain: new mongoose.Types.ObjectId(token.domainId), + commentId: token.commentId, + }).lean(); + + const replyObj = { + domain: new mongoose.Types.ObjectId(token.domainId), + productId: token.entityId, + entityType: "lesson", + entityId: parentComment?.entityId || token.entityId, + commentId: token.commentId, + replyId: crypto.randomUUID(), + parentReplyId: token.parentReplyId || undefined, + userId, + content: [{ type: "paragraph", children: [{ text: content }] }], + likesCount: 0, + deleted: false, + isEdited: false, + }; + + await ProductDiscussionReplyModel.create(replyObj); + + await dispatchReplyNotification({ + domain: token.domainId, + userId, + activityType: "course_discussion_comment_created", + entityId: replyObj.replyId, + metadata: { + courseId: token.entityId, + entityType: "lesson", + entityId: replyObj.entityId, + commentId: token.commentId, + replyId: replyObj.replyId, + eventType: "reply_created", + contentType: "reply", + commentContent: content, + }, + }); +} + +/** + * Dispatch notification to subscribers when a reply comes in via email. + */ +async function dispatchReplyNotification(params: { + domain: string; + userId: string; + activityType: string; + entityId: string; + entityTargetId?: string; + metadata?: Record; +}): Promise { + const { addDispatchNotificationJob } = await import( + "../notifications/services/enqueue" + ); + + await addDispatchNotificationJob({ + domain: new mongoose.Types.ObjectId(params.domain), + userId: params.userId, + activityType: params.activityType as any, + entityId: params.entityId, + entityTargetId: params.entityTargetId, + metadata: params.metadata || {}, + }); +} diff --git a/apps/queue/src/inbound-email/provider.ts b/apps/queue/src/inbound-email/provider.ts new file mode 100644 index 000000000..e5f794ea6 --- /dev/null +++ b/apps/queue/src/inbound-email/provider.ts @@ -0,0 +1,315 @@ +/** + * Vendor-agnostic inbound email abstraction. + * + * CourseLit can be configured to receive inbound replies via: + * - AWS SES (SNS notification) + * - Postmark (Inbound webhook) + * - Mailgun (Route + webhook) + * - SendGrid (Inbound Parse webhook) + * + * All providers normalize to a common `InboundEmailPayload` before processing. + */ + +/** + * Normalized payload from any inbound email provider. + */ +export interface InboundEmailPayload { + /** The sender's email address (the person replying) */ + from: string; + /** The recipient address (our reply-to address, contains the token) */ + to: string; + /** Plain text body of the email (stripped of quoted content later) */ + text: string; + /** HTML body of the email (optional) */ + html?: string; + /** The original subject line */ + subject?: string; + /** Message ID of the inbound email */ + messageId?: string; + /** Provider-specific raw data for debugging */ + raw?: Record; +} + +/** + * Interface for inbound email providers. + * Each provider implements this to parse its webhook payload into our normalized format. + */ +export interface InboundEmailProvider { + /** Provider name identifier */ + name: string; + /** + * Parse the incoming webhook body into a normalized InboundEmailPayload. + * Returns null if the payload is invalid or cannot be parsed. + */ + parse(payload: Record): InboundEmailPayload | null; +} + +/** + * AWS SES provider implementation. + * SES delivers inbound emails via SNS notifications. + */ +export class SESInboundProvider implements InboundEmailProvider { + name = "ses"; + + parse(payload: Record): InboundEmailPayload | null { + try { + // SES SNS notification wraps the mail content + const snsMessage = + typeof payload.Message === "string" + ? JSON.parse(payload.Message) + : payload.Message; + + const mail = snsMessage?.mail || snsMessage; + const content = snsMessage?.content || mail?.commonHeaders; + + if (!mail) { + return null; + } + + // SES sends the email body as base64 in the `content` field + const rawText = + typeof snsMessage?.content === "string" + ? Buffer.from(snsMessage.content, "base64").toString( + "utf-8", + ) + : ""; + + // Extract text/plain part + const text = extractPlainTextFromRawEmail(rawText); + + return { + from: extractFirstAddress( + content?.from || mail?.commonHeaders?.from?.[0] || "", + ), + to: extractFirstAddress( + content?.to || mail?.commonHeaders?.to?.[0] || "", + ), + text: text || "", + html: rawText, + subject: content?.subject || mail?.commonHeaders?.subject, + messageId: mail?.messageId, + raw: payload, + }; + } catch { + return null; + } + } +} + +/** + * Postmark provider implementation. + */ +export class PostmarkInboundProvider implements InboundEmailProvider { + name = "postmark"; + + parse(payload: Record): InboundEmailPayload | null { + if (!payload.From || !payload.To) { + return null; + } + + return { + from: String(payload.From), + to: String(payload.To), + text: String(payload.TextBody || payload.StrippedTextReply || ""), + html: String(payload.HtmlBody || ""), + subject: payload.Subject ? String(payload.Subject) : undefined, + messageId: payload.MessageID + ? String(payload.MessageID) + : undefined, + raw: payload, + }; + } +} + +/** + * Mailgun provider implementation. + */ +export class MailgunInboundProvider implements InboundEmailProvider { + name = "mailgun"; + + parse(payload: Record): InboundEmailPayload | null { + if (!payload.from || !payload.recipient) { + return null; + } + + return { + from: String(payload.from), + to: String(payload.recipient), + text: String(payload["body-plain"] || payload["stripped-text"] || ""), + html: String(payload["body-html"] || ""), + subject: payload.subject ? String(payload.subject) : undefined, + messageId: payload["message-id"] + ? String(payload["message-id"]) + : undefined, + raw: payload, + }; + } +} + +/** + * SendGrid Inbound Parse provider implementation. + */ +export class SendGridInboundProvider implements InboundEmailProvider { + name = "sendgrid"; + + parse(payload: Record): InboundEmailPayload | null { + if (!payload.from || !payload.to) { + return null; + } + + return { + from: String(payload.from), + to: String(payload.to), + text: String(payload.text || payload["stripped-text"] || ""), + html: String(payload.html || payload["stripped-html"] || ""), + subject: payload.subject ? String(payload.subject) : undefined, + messageId: payload.message_id + ? String(payload.message_id) + : undefined, + raw: payload, + }; + } +} + +/** + * Factory to get the right provider based on config. + */ +export function getInboundEmailProvider( + providerName?: string, +): InboundEmailProvider { + const name = + providerName || + process.env.INBOUND_EMAIL_PROVIDER || + "ses"; + + switch (name) { + case "postmark": + return new PostmarkInboundProvider(); + case "mailgun": + return new MailgunInboundProvider(); + case "sendgrid": + return new SendGridInboundProvider(); + case "ses": + default: + return new SESInboundProvider(); + } +} + +/** + * Extract the email address from a string like "Name " + * or just "email@example.com". + */ +function extractFirstAddress(address: string): string { + const match = address.match(/<([^>]+)>/) || address.match(/([^\s,;]+)/); + return match ? match[1] || match[0] : address; +} + +/** + * Crude plain-text extraction from raw email content (MIME). + * Uses line-by-line extraction of text/plain parts. + */ +function extractPlainTextFromRawEmail(raw: string): string { + if (!raw) return ""; + + // Try to find text/plain part in multipart MIME + const textPlainMatch = raw.match( + /Content-Type:\s*text\/plain[\s\S]*?(?:\r?\n\r?\n)([\s\S]*?)(?:\r?\n--|$)/i, + ); + if (textPlainMatch) { + return textPlainMatch[1].trim(); + } + + // If no MIME parsing works, return raw (it might already be plain text) + return raw.trim(); +} + +/** + * Strips quoted/replied content from an email body. + * Handles common patterns: + * - "On ... wrote:" blocks + * - "> " quoted lines + * - "--- Original Message ---" separators + * - "From: ..." forwards + */ +export function stripQuotedContent(text: string): string { + if (!text) return ""; + + const lines = text.split("\n"); + const result: string[] = []; + + // Patterns that indicate quoted content start + const quoteStartPatterns = [ + /^On\s.+\s?wrote:\s*$/i, + /^-{3,}\s*Original\s*Message\s*-{3,}\s*$/i, + /^_{3,}\s*Original\s*Message\s*_{3,}\s*$/i, + /^>+\s*On\s.+\s?wrote:\s*$/i, + /^From:.*$/i, + /^Sent:\s.*$/i, + /^To:.*$/i, + /^Subject:.*$/i, + /^Date:.*$/i, + /^Reply\s*above\s*this\s*line/i, + /^#{3,}\s*Forwarded\s*message\s*#{3,}/i, + ]; + + let inQuotedSection = false; + let consecutiveQuotedLines = 0; + + for (const line of lines) { + const trimmed = line.trim(); + + // Check if this line starts a quoted section + const isQuoteMarker = quoteStartPatterns.some((p) => + p.test(trimmed), + ); + + if (isQuoteMarker) { + inQuotedSection = true; + continue; + } + + // Check for "> " quoted lines + const isQuotedLine = /^>/.test(line); + + if (inQuotedSection) { + // Keep going until we hit a non-quoted section + if (trimmed === "" && consecutiveQuotedLines > 0) { + // Empty lines inside quoted section + continue; + } + if (isQuotedLine) { + consecutiveQuotedLines++; + continue; + } + if (trimmed === "") { + continue; + } + // Check if this is an email signature + if (/^--\s*$/.test(trimmed)) { + inQuotedSection = true; + continue; + } + // If we hit content that's not quoted, we're past the quoted section + if (consecutiveQuotedLines > 2) { + // We're still in quoted territory, skip + continue; + } + inQuotedSection = false; + } + + if (isQuotedLine) { + // Count consecutive quoted lines + consecutiveQuotedLines++; + if (consecutiveQuotedLines > 3) { + // Likely a quoted block + continue; + } + } else { + consecutiveQuotedLines = 0; + } + + result.push(line); + } + + return result.join("\n").trim(); +} diff --git a/apps/queue/src/inbound-email/routes.ts b/apps/queue/src/inbound-email/routes.ts new file mode 100644 index 000000000..5288f8440 --- /dev/null +++ b/apps/queue/src/inbound-email/routes.ts @@ -0,0 +1,75 @@ +import express from "express"; +import { getInboundEmailProvider } from "./provider"; +import { processInboundReply } from "./processor"; +import { logger } from "../logger"; +import { captureError, getDomainId } from "../observability/posthog"; + +const router: any = express.Router(); + +/** + * POST /inbound-email + * + * Universal inbound email webhook endpoint. + * Accepts normalized payloads from any supported provider (SES, Postmark, Mailgun, SendGrid). + * The provider auto-detection is based on the shape of the incoming JSON payload. + */ +router.post("/", async (req: express.Request, res: express.Response) => { + try { + const provider = getInboundEmailProvider( + req.query.provider as string | undefined, + ); + + const parsed = provider.parse(req.body); + if (!parsed) { + logger.warn( + { provider: provider.name }, + "Inbound email could not be parsed", + ); + // Return 200 to prevent provider from retrying (bad payload) + return res.status(200).json({ success: false, error: "Unparseable payload" }); + } + + logger.info( + { + provider: provider.name, + from: parsed.from, + to: parsed.to, + subject: parsed.subject, + }, + "Processing inbound email reply", + ); + + const result = await processInboundReply(parsed); + + if (!result.success) { + logger.warn( + { error: result.error, from: parsed.from }, + "Inbound email processing failed", + ); + } else { + logger.info( + { from: parsed.from }, + "Inbound email reply processed successfully", + ); + } + + // Always return 200 to prevent provider retries + return res.status(200).json(result); + } catch (err: any) { + logger.error(err, "Inbound email route error"); + captureError({ + error: err, + source: "route.inbound_email", + domainId: getDomainId(), + context: { + path: req.path, + method: req.method, + route: "/inbound-email", + }, + }); + // Return 200 to prevent provider retries + return res.status(200).json({ success: false, error: "Internal error" }); + } +}); + +export default router; diff --git a/apps/queue/src/index.ts b/apps/queue/src/index.ts index 8ffd4ae3f..d060b8675 100644 --- a/apps/queue/src/index.ts +++ b/apps/queue/src/index.ts @@ -1,6 +1,7 @@ import express from "express"; import jobRoutes from "./job/routes"; import sseRoutes from "./sse/routes"; +import inboundEmailRoutes from "./inbound-email/routes"; // start workers import "./domain/worker"; @@ -23,6 +24,7 @@ app.use(express.json()); app.use("/job", verifyJWTMiddleware, jobRoutes); app.use("/sse", sseRoutes); +app.use("/inbound-email", inboundEmailRoutes); app.get("/healthy", (req, res) => { res.status(200).json({ status: "ok", uptime: process.uptime() }); diff --git a/apps/queue/src/notifications/services/channels/email.ts b/apps/queue/src/notifications/services/channels/email.ts index e98afdb17..c42320386 100644 --- a/apps/queue/src/notifications/services/channels/email.ts +++ b/apps/queue/src/notifications/services/channels/email.ts @@ -1,4 +1,9 @@ -import { getNotificationMessageAndHref } from "@courselit/common-logic"; +import { + getRicherNotificationContent, + getNotificationMessageAndHref, + createReplyToken, + buildReplyToAddress, +} from "@courselit/common-logic"; import { renderEmailToHtml } from "@courselit/email-editor"; import { getEmailFrom } from "@courselit/utils"; import { addMailJob } from "../../../domain/handler"; @@ -6,12 +11,88 @@ import { getSiteUrl } from "../../../utils/get-site-url"; import { getUnsubLink } from "../../../utils/get-unsub-link"; import { ChannelPayload, NotificationChannel } from "./types"; import { getDomainId } from "../../../observability/posthog"; +import { buildRicherNotificationEmailTemplate } from "./richer-notification-email-template"; import { buildNotificationEmailTemplate } from "./notification-email-template"; +import type { ReplyTokenPayload } from "@courselit/common-models"; function getActorAvatarUrl(actor: ChannelPayload["actor"]) { return actor?.avatar?.file || actor?.avatar?.thumbnail || undefined; } +/** + * Determines if a given activity type is a "discussion" type that should get + * richer email content and reply-by-email support. + */ +function isDiscussionActivityType(activityType: string): boolean { + const discussionTypes = [ + "community_post_created", + "community_comment_created", + "community_comment_replied", + "community_reply_created", + "course_discussion_comment_created", + "course_discussion_reacted", + ]; + return discussionTypes.includes(activityType); +} + +/** + * Builds a ReplyTokenPayload for discussion activity types so that + * the recipient can reply directly from their email client. + */ +function buildReplyTokenPayload( + payload: ChannelPayload, +): ReplyTokenPayload | null { + const { activityType, entityId, entityTargetId, metadata, domain, actorUserId } = payload; + + switch (activityType) { + case "community_comment_created": { + const postId = metadata?.postId as string; + return { + userId: actorUserId, + domainId: String(domain?._id), + entityId: postId || entityId, + entityType: "community", + commentId: entityId, + }; + } + case "community_reply_created": + case "community_comment_replied": { + const commentId = entityTargetId || (metadata?.commentId as string); + return { + userId: actorUserId, + domainId: String(domain?._id), + entityId: entityId, + entityType: "community", + commentId, + parentReplyId: entityId, + }; + } + case "community_post_created": { + return { + userId: actorUserId, + domainId: String(domain?._id), + entityId: entityId, + entityType: "community", + }; + } + case "course_discussion_comment_created": + case "course_discussion_reacted": { + const commentId = metadata?.commentId as string; + const replyId = metadata?.replyId as string; + return { + userId: actorUserId, + domainId: String(domain?._id), + entityId: (metadata?.courseId as string) || entityId, + entityType: "product", + commentId: commentId || entityId, + parentReplyId: replyId || undefined, + }; + } + default: + return null; + } +} + export class EmailChannel implements NotificationChannel { async send(payload: ChannelPayload): Promise { if (!payload.recipient.email || !payload.recipient.unsubscribeToken) { @@ -27,6 +108,91 @@ export class EmailChannel implements NotificationChannel { payload.actor?.email || payload.actor?.userId || "Someone"; + + const isDiscussion = isDiscussionActivityType(payload.activityType); + + if (isDiscussion) { + await this.sendRicherNotification(payload, actorName); + } else { + await this.sendLegacyNotification(payload, actorName); + } + } + + private async sendRicherNotification( + payload: ChannelPayload, + actorName: string, + ): Promise { + const notificationContent = await getRicherNotificationContent({ + activityType: payload.activityType, + entityId: payload.entityId, + actorName, + recipientUserId: payload.recipient.userId, + recipientPermissions: payload.recipient.permissions || [], + entityTargetId: payload.entityTargetId, + metadata: payload.metadata, + hrefPrefix: getSiteUrl(payload.domain), + domainId: payload.domain?._id, + }); + + if (!notificationContent.message || !notificationContent.href) { + return; + } + + const unsubscribeUrl = getUnsubLink( + payload.domain, + payload.recipient.unsubscribeToken, + ); + + // Build reply-to address for discussion-type notifications + let replyToAddress: string | undefined; + try { + const replyTokenPayload = buildReplyTokenPayload(payload); + if (replyTokenPayload) { + const token = createReplyToken(replyTokenPayload); + replyToAddress = buildReplyToAddress(token); + } + } catch { + // If reply token creation fails, still send the notification without reply-to + } + + const body = await renderEmailToHtml({ + email: buildRicherNotificationEmailTemplate({ + content: notificationContent, + actorName, + actorAvatarUrl: getActorAvatarUrl(payload.actor), + unsubscribeUrl, + hideCourseLitBranding: + payload.domain.settings?.hideCourseLitBranding, + replyToAddress, + }), + }); + + const headers: Record = { + "List-Unsubscribe": `<${unsubscribeUrl}>`, + "List-Unsubscribe-Post": "List-Unsubscribe=One-Click", + }; + + if (replyToAddress) { + headers["Reply-To"] = replyToAddress; + } + + await addMailJob({ + to: [payload.recipient.email], + from: getEmailFrom({ + name: payload.domain.settings?.title || payload.domain.name, + email: process.env.EMAIL_FROM || "", + }), + domainId: getDomainId(payload.domain?._id), + subject: notificationContent.subject || notificationContent.message, + body, + headers, + }); + } + + private async sendLegacyNotification( + payload: ChannelPayload, + actorName: string, + ): Promise { const notificationDetails = await getNotificationMessageAndHref({ activityType: payload.activityType, entityId: payload.entityId, @@ -47,6 +213,7 @@ export class EmailChannel implements NotificationChannel { payload.domain, payload.recipient.unsubscribeToken, ); + const body = await renderEmailToHtml({ email: buildNotificationEmailTemplate({ actorName, diff --git a/apps/queue/src/notifications/services/channels/richer-notification-email-template.ts b/apps/queue/src/notifications/services/channels/richer-notification-email-template.ts new file mode 100644 index 000000000..0504955ca --- /dev/null +++ b/apps/queue/src/notifications/services/channels/richer-notification-email-template.ts @@ -0,0 +1,306 @@ +import type { Email } from "@courselit/email-editor"; +import type { RicherNotificationContent } from "@courselit/common-logic"; + +interface RicherNotificationEmailInput { + content: RicherNotificationContent; + actorName: string; + actorAvatarUrl?: string; + unsubscribeUrl: string; + hideCourseLitBranding?: boolean; + replyToAddress?: string; +} + +function encodePlainTextForMarkdown(value: string) { + return Array.from(value) + .map((character) => `&#${character.codePointAt(0)};`) + .join(""); +} + +function getSafeAvatarUrl(actorAvatarUrl?: string) { + if (!actorAvatarUrl) return; + if (actorAvatarUrl.startsWith("/")) return actorAvatarUrl; + try { + const url = new URL(actorAvatarUrl); + if (url.protocol === "https:" || url.protocol === "http:") { + return actorAvatarUrl; + } + } catch { + return; + } +} + +export function buildRicherNotificationEmailTemplate({ + content, + actorName, + actorAvatarUrl, + unsubscribeUrl, + hideCourseLitBranding, + replyToAddress, +}: RicherNotificationEmailInput): Email { + const safeActorAvatarUrl = getSafeAvatarUrl(actorAvatarUrl); + const emailContent: Email["content"] = []; + + // --- Header / Actor section --- + if (safeActorAvatarUrl) { + emailContent.push({ + blockType: "image", + settings: { + src: safeActorAvatarUrl, + alt: `${actorName} avatar`, + alignment: "left", + width: "40px", + height: "40px", + maxWidth: "40px", + borderRadius: "999px", + paddingTop: "24px", + paddingBottom: "10px", + paddingX: "24px", + }, + }); + } + + emailContent.push({ + blockType: "text", + settings: { + content: `**${encodePlainTextForMarkdown(actorName)}**`, + fontSize: "14px", + lineHeight: "1.4", + paddingTop: safeActorAvatarUrl ? "0px" : "24px", + paddingBottom: "4px", + }, + }); + + // --- Notification message line --- + emailContent.push({ + blockType: "text", + settings: { + content: encodePlainTextForMarkdown(content.message), + fontSize: "16px", + lineHeight: "1.6", + paddingTop: "8px", + paddingBottom: "12px", + }, + }); + + // --- Separator --- + emailContent.push({ + blockType: "separator", + settings: { + paddingTop: "0px", + paddingBottom: "0px", + }, + }); + + // --- Content details section --- + if (content.discussionTitle) { + emailContent.push({ + blockType: "text", + settings: { + content: `**Discussion:** ${encodePlainTextForMarkdown(content.discussionTitle)}`, + fontSize: "14px", + lineHeight: "1.5", + paddingTop: "16px", + paddingBottom: "8px", + }, + }); + } + + if (content.parentCommentContent) { + emailContent.push({ + blockType: "text", + settings: { + content: `> ${encodePlainTextForMarkdown(content.parentCommentContent)}`, + fontSize: "13px", + lineHeight: "1.5", + foregroundColor: "#666666", + paddingTop: "4px", + paddingBottom: "8px", + }, + }); + } + + if (content.commentContent) { + emailContent.push({ + blockType: "text", + settings: { + content: encodePlainTextForMarkdown(content.commentContent), + fontSize: "15px", + lineHeight: "1.6", + paddingTop: "4px", + paddingBottom: "16px", + }, + }); + } + + if (content.postContent) { + emailContent.push({ + blockType: "text", + settings: { + content: encodePlainTextForMarkdown(content.postContent), + fontSize: "14px", + lineHeight: "1.5", + foregroundColor: "#444444", + paddingTop: "4px", + paddingBottom: "16px", + }, + }); + } + + if (content.communityName) { + emailContent.push({ + blockType: "text", + settings: { + content: `*In ${encodePlainTextForMarkdown(content.communityName)}*`, + fontSize: "12px", + lineHeight: "1.4", + foregroundColor: "#888888", + paddingTop: "0px", + paddingBottom: "16px", + }, + }); + } + + if (content.courseTitle) { + emailContent.push({ + blockType: "text", + settings: { + content: `*From course: ${encodePlainTextForMarkdown(content.courseTitle)}*`, + fontSize: "12px", + lineHeight: "1.4", + foregroundColor: "#888888", + paddingTop: "0px", + paddingBottom: "16px", + }, + }); + } + + // --- CTA Button --- + emailContent.push({ + blockType: "link", + settings: { + text: "View in CourseLit", + url: content.href, + alignment: "center", + isButton: true, + buttonColor: "#000000", + buttonTextColor: "#ffffff", + buttonBorderRadius: "6px", + buttonPaddingX: "18px", + buttonPaddingY: "10px", + paddingTop: "12px", + paddingBottom: "24px", + }, + }); + + // --- Reply hint if reply-to is set --- + if (replyToAddress) { + emailContent.push({ + blockType: "text", + settings: { + content: "💬 Reply to this email to add a comment or reply to the discussion.", + alignment: "center", + fontSize: "12px", + foregroundColor: "#888888", + paddingTop: "8px", + paddingBottom: "16px", + }, + }); + } + + // --- Separator + Footer --- + emailContent.push({ + blockType: "separator", + settings: { + paddingTop: "0px", + paddingBottom: "0px", + }, + }); + + emailContent.push({ + blockType: "text", + settings: { + content: `[Unsubscribe from email notifications](${unsubscribeUrl})`, + alignment: "center", + fontSize: "12px", + foregroundColor: "#666666", + paddingTop: "32px", + paddingBottom: hideCourseLitBranding ? "32px" : "16px", + }, + }); + + if (!hideCourseLitBranding) { + emailContent.push({ + blockType: "link", + settings: { + text: "Powered by CourseLit", + url: "https://courselit.app", + alignment: "center", + fontSize: "12px", + textColor: "#000000", + textDecoration: "none", + buttonBorderColor: "#000000", + paddingTop: "0px", + paddingBottom: "40px", + }, + }); + } + + return { + style: { + colors: { + background: "#f7f7f7", + foreground: "#000000", + border: "#e5e5e5", + accent: "#000000", + accentForeground: "#ffffff", + }, + typography: { + header: { + fontFamily: "Arial, sans-serif", + letterSpacing: "0px", + textTransform: "none", + textDecoration: "none", + }, + text: { + fontFamily: "Arial, sans-serif", + letterSpacing: "0px", + textTransform: "none", + textDecoration: "none", + }, + link: { + fontFamily: "Arial, sans-serif", + letterSpacing: "0px", + textTransform: "none", + textDecoration: "underline", + }, + }, + interactives: { + button: { + padding: { x: "18px", y: "10px" }, + border: { width: "0px", radius: "6px", style: "solid" }, + }, + link: { + padding: { x: "0px", y: "0px" }, + }, + }, + structure: { + page: { + background: "#ffffff", + foreground: "#000000", + width: "600px", + marginY: "20px", + borderWidth: "1px", + borderStyle: "solid", + borderRadius: "8px", + }, + section: { + padding: { x: "24px", y: "16px" }, + }, + }, + }, + meta: { + previewText: content.commentContent || content.postContent || content.message, + }, + content: emailContent, + }; +} diff --git a/apps/web/app/api/inbound-email/route.ts b/apps/web/app/api/inbound-email/route.ts new file mode 100644 index 000000000..45f44f6f5 --- /dev/null +++ b/apps/web/app/api/inbound-email/route.ts @@ -0,0 +1,61 @@ +import { NextRequest, NextResponse } from "next/server"; + +/** + * POST /api/inbound-email + * + * Receives inbound email webhooks from email providers (SES, Postmark, Mailgun, SendGrid). + * Proxies the request to the queue service for processing. + * + * The queue service (apps/queue) handles all the business logic: + * - Parsing the provider-specific payload + * - Verifying reply tokens + * - Stripping quoted content + * - Creating replies in the database + * - Dispatching notifications + */ +export async function POST(req: NextRequest) { + try { + const body = await req.json(); + const provider = req.nextUrl.searchParams.get("provider") || undefined; + + const queueUrl = process.env.QUEUE_SERVICE_URL; + if (!queueUrl) { + console.warn( + "QUEUE_SERVICE_URL not set, cannot process inbound email. Set the environment variable to enable reply-by-email.", + ); + return NextResponse.json( + { + success: false, + error: "Inbound email processing not configured", + }, + { status: 200 }, + ); + } + + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), 15000); + + const response = await fetch( + `${queueUrl}/inbound-email${provider ? `?provider=${provider}` : ""}`, + { + method: "POST", + headers: { + "Content-Type": "application/json", + }, + body: JSON.stringify(body), + signal: controller.signal, + }, + ); + + clearTimeout(timeout); + + const result = await response.json(); + return NextResponse.json(result, { status: 200 }); + } catch (err: any) { + console.error("Inbound email proxy error:", err); + return NextResponse.json( + { success: false, error: "Internal error" }, + { status: 200 }, + ); + } +} diff --git a/packages/common-logic/src/index.ts b/packages/common-logic/src/index.ts index 1c01ac737..cf7d2648e 100644 --- a/packages/common-logic/src/index.ts +++ b/packages/common-logic/src/index.ts @@ -2,3 +2,5 @@ export * from "./utils/convert-filters-to-db-conditions"; export * from "./utils/course-management-access"; export * from "./utils/get-notification-message-and-href"; export { default as jwtUtils } from "./utils/jwt-utils"; +export * from "./utils/reply-token-utils"; +export * from "./utils/get-richer-notification-content"; diff --git a/packages/common-logic/src/utils/get-richer-notification-content.ts b/packages/common-logic/src/utils/get-richer-notification-content.ts new file mode 100644 index 000000000..1d85244fe --- /dev/null +++ b/packages/common-logic/src/utils/get-richer-notification-content.ts @@ -0,0 +1,225 @@ +import { Constants, ActivityType } from "@courselit/common-models"; +import { truncate, extractTextFromTextEditorContent } from "@courselit/utils"; +import { createNotificationEntityResolver } from "./notification-entity-resolver"; +import type { NotificationEntityResolver, NotificationCommentEntity, NotificationPostEntity, NotificationCommunityEntity, NotificationCourseEntity } from "./get-notification-message-and-href"; + +export interface RicherNotificationContent { + message: string; + href: string; + subject: string; + authorName: string; + discussionTitle?: string; + parentCommentContent?: string; + commentContent?: string; + postContent?: string; + communityName?: string; + courseTitle?: string; + entityType: "community_post" | "community_comment" | "community_reply" | "course_discussion_comment" | "course_discussion_reply" | "other"; +} + +const defaultEntityResolver = createNotificationEntityResolver(); + +function extractContentString(content: unknown): string { + if (!content) return ""; + if (typeof content === "string") return content; + if (Array.isArray(content)) { + return extractTextFromTextEditorContent(content as any) || ""; + } + return String(content); +} + +export async function getRicherNotificationContent({ + activityType, + entityId, + actorName, + recipientUserId, + recipientPermissions = [], + entityTargetId, + metadata, + hrefPrefix = "", + domainId, + resolver, +}: { + activityType: ActivityType; + entityId: string; + actorName: string; + recipientUserId: string; + recipientPermissions?: string[]; + entityTargetId?: string; + metadata?: Record; + hrefPrefix?: string; + domainId?: unknown; + resolver?: NotificationEntityResolver; +}): Promise { + const entityResolver = resolver || defaultEntityResolver; + + const base: RicherNotificationContent = { + message: "", + href: "", + subject: "", + authorName: actorName, + entityType: "other", + }; + + switch (activityType) { + case Constants.ActivityType.COMMUNITY_POST_CREATED: { + const post = await entityResolver.getPost(entityId, domainId); + if (!post) return base; + + const community = await entityResolver.getCommunity(post.communityId, domainId); + if (!community) return base; + + const postContentStr = extractContentString(metadata?.postContent); + + return { + ...base, + message: `${actorName} created a post '${truncate(post.title, 60).trim()}' in ${community.name}`, + subject: `New post: ${post.title}`, + href: toHref(`/dashboard/community/${community.communityId}/${post.postId}`, hrefPrefix), + discussionTitle: post.title, + postContent: postContentStr || undefined, + communityName: community.name, + entityType: "community_post", + }; + } + + case Constants.ActivityType.COMMUNITY_COMMENT_CREATED: { + const postId = (metadata?.postId as string) || (await entityResolver.getComment(entityId, domainId))?.postId || entityId; + const post = await entityResolver.getPost(postId, domainId); + if (!post) return base; + + const community = await entityResolver.getCommunity(post.communityId, domainId); + if (!community) return base; + + const comment = await entityResolver.getComment(entityId, domainId); + + return { + ...base, + message: `${actorName} commented on ${recipientUserId === post.userId ? "your" : "a"} post '${truncate(post.title, 60).trim()}' in ${community.name}`, + subject: `Re: ${post.title}`, + href: toHref(`/dashboard/community/${community.communityId}/${post.postId}#${entityId}`, hrefPrefix), + discussionTitle: post.title, + commentContent: comment?.content ? truncate(comment.content, 200).trim() : undefined, + communityName: community.name, + entityType: "community_comment", + }; + } + + case Constants.ActivityType.COMMUNITY_REPLY_CREATED: + case Constants.ActivityType.COMMUNITY_COMMENT_REPLIED: { + const commentId = entityTargetId || (metadata?.commentId as string) || ""; + if (!commentId) return base; + + const comment = await entityResolver.getComment(commentId, domainId); + if (!comment) return base; + + const reply = comment.replies.find((r) => r.replyId === entityId); + if (!reply) return base; + + const parentReply = reply.parentReplyId + ? comment.replies.find((r) => r.replyId === reply.parentReplyId) + : undefined; + + const [post, community] = await Promise.all([ + entityResolver.getPost(comment.postId, domainId), + entityResolver.getCommunity(comment.communityId, domainId), + ]); + if (!post || !community) return base; + + const prefix = parentReply + ? recipientUserId === parentReply.userId ? "your" : "a" + : recipientUserId === comment.userId ? "your" : "a"; + + return { + ...base, + message: `${actorName} replied to ${prefix} comment on '${truncate(post.title, 60).trim()}' in ${community.name}`, + subject: `Re: ${post.title}`, + href: toHref(`/dashboard/community/${community.communityId}/${post.postId}#${entityId}`, hrefPrefix), + discussionTitle: post.title, + parentCommentContent: parentReply + ? truncate(parentReply.content, 200).trim() + : comment.content + ? truncate(comment.content, 200).trim() + : undefined, + commentContent: truncate(reply.content, 200).trim(), + communityName: community.name, + entityType: "community_reply", + }; + } + + case Constants.ActivityType.COURSE_DISCUSSION_COMMENT_CREATED: + case Constants.ActivityType.COURSE_DISCUSSION_REACTED: { + const productId = metadata?.courseId as string | undefined; + const entityType = metadata?.entityType as string | undefined; + const lessonId = metadata?.entityId as string | undefined; + const commentId = metadata?.commentId as string | undefined; + const replyId = metadata?.replyId as string | undefined; + const eventType = metadata?.eventType as string | undefined; + const contentType = metadata?.contentType as string | undefined; + const commentContent = metadata?.commentContent as string | undefined; + const parentCommentContent = metadata?.parentCommentContent as string | undefined; + + if (!productId || entityType !== Constants.ProductDiscussionEntityType.LESSON || !lessonId) { + return base; + } + + const course = await entityResolver.getCourse(productId, domainId); + if (!course?.slug) return base; + + const targetId = replyId || commentId; + const targetHash = replyId + ? `discussion-reply-${replyId}` + : commentId + ? `discussion-comment-${commentId}` + : undefined; + const query = new URLSearchParams({ discussion: "open" }); + + // Import would create circular dep, so inline the check + const canManage = String(course.creatorId) === recipientUserId || + recipientPermissions.includes("course:manage_any"); + if (canManage) { + query.set("preview", "true"); + } + + const isReply = eventType === "reply_created" || contentType === Constants.ProductDiscussionContentType.REPLY; + const isReact = activityType === Constants.ActivityType.COURSE_DISCUSSION_REACTED; + + return { + ...base, + message: isReact + ? `${actorName} reacted to your ${contentType === Constants.ProductDiscussionContentType.REPLY ? "reply" : "comment"} on ${truncate(course.title, 60).trim()}` + : `${actorName} ${isReply ? "replied" : "commented"} on ${truncate(course.title, 60).trim()}`, + subject: isReact + ? `Reaction on: ${course.title}` + : `Re: ${course.title}`, + href: toHref( + `/course/${course.slug}/${course.courseId}/${lessonId}?${query.toString()}${targetId && targetHash ? `#${targetHash}` : ""}`, + hrefPrefix, + ), + discussionTitle: course.title, + commentContent: commentContent || undefined, + parentCommentContent: parentCommentContent || undefined, + courseTitle: course.title, + entityType: isReply ? "course_discussion_reply" : "course_discussion_comment", + }; + } + + default: { + return { + ...base, + message: `${actorName} triggered ${humanizeActivityType(activityType)}`, + subject: `Notification: ${actorName}`, + href: toHref("/dashboard", hrefPrefix), + }; + } + } +} + +function humanizeActivityType(activityType: string): string { + return activityType.replace(/_/g, " "); +} + +function toHref(path: string, hrefPrefix: string): string { + if (!hrefPrefix) return path; + return `${hrefPrefix.replace(/\/$/, "")}${path}`; +} diff --git a/packages/common-logic/src/utils/reply-token-utils.ts b/packages/common-logic/src/utils/reply-token-utils.ts new file mode 100644 index 000000000..ea9dbfe2f --- /dev/null +++ b/packages/common-logic/src/utils/reply-token-utils.ts @@ -0,0 +1,114 @@ +import crypto from "crypto"; +import type { ReplyTokenPayload } from "@courselit/common-models"; + +const ALGORITHM = "aes-256-gcm"; +const IV_LENGTH = 16; +const AUTH_TAG_LENGTH = 16; + +function getEncryptionKey(): Buffer { + const key = process.env.REPLY_EMAIL_SECRET; + if (!key) { + throw new Error("REPLY_EMAIL_SECRET environment variable is not set"); + } + // Ensure key is 32 bytes for AES-256 + return crypto.scryptSync(key, "courselit-reply-salt", 32); +} + +function encodeBase64UrlSafe(buf: Buffer): string { + return buf + .toString("base64") + .replace(/\+/g, "-") + .replace(/\//g, "_") + .replace(/=+$/, ""); +} + +function decodeBase64UrlSafe(str: string): Buffer { + const padded = str.replace(/-/g, "+").replace(/_/g, "/"); + return Buffer.from(padded, "base64"); +} + +/** + * Creates a signed, encrypted reply token that can be embedded in the Reply-To header + * of notification emails. The token encodes the user, domain, discussion context, and + * a timestamp for expiration. + */ +export function createReplyToken(payload: ReplyTokenPayload): string { + const key = getEncryptionKey(); + const iv = crypto.randomBytes(IV_LENGTH); + + const data = JSON.stringify({ + ...payload, + iat: Math.floor(Date.now() / 1000), + exp: Math.floor(Date.now() / 1000) + 30 * 24 * 60 * 60, // 30 days + }); + + const cipher = crypto.createCipheriv(ALGORITHM, key, iv); + const encrypted = Buffer.concat([ + cipher.update(data, "utf8"), + cipher.final(), + ]); + const authTag = cipher.getAuthTag(); + + // Format: iv:authTag:encryptedData (all base64url safe) + return [ + encodeBase64UrlSafe(iv), + encodeBase64UrlSafe(authTag), + encodeBase64UrlSafe(encrypted), + ].join(":"); +} + +/** + * Decrypts and validates a reply token. Returns the payload if valid, or null if + * the token is invalid, expired, or tampered with. + */ +export function verifyReplyToken(token: string): ReplyTokenPayload | null { + try { + const parts = token.split(":"); + if (parts.length !== 3) { + return null; + } + + const iv = decodeBase64UrlSafe(parts[0]); + const authTag = decodeBase64UrlSafe(parts[1]); + const encrypted = decodeBase64UrlSafe(parts[2]); + + const key = getEncryptionKey(); + const decipher = crypto.createDecipheriv(ALGORITHM, key, iv); + decipher.setAuthTag(authTag); + + const decrypted = Buffer.concat([ + decipher.update(encrypted), + decipher.final(), + ]); + + const payload = JSON.parse(decrypted.toString("utf8")) as ReplyTokenPayload & { + iat: number; + exp: number; + }; + + // Check expiration + if (payload.exp < Math.floor(Date.now() / 1000)) { + return null; + } + + const { iat, exp, ...rest } = payload; + return rest as ReplyTokenPayload; + } catch { + return null; + } +} + +/** + * Extracts the reply-from email address to use as the Reply-To header. + * Format: reply+@ + */ +export function buildReplyToAddress( + token: string, + inboundDomain?: string, +): string { + const domain = inboundDomain || process.env.INBOUND_EMAIL_DOMAIN || ""; + if (!domain) { + return ""; + } + return `reply+${token}@${domain}`; +} diff --git a/packages/common-models/src/index.ts b/packages/common-models/src/index.ts index 6ed6f778f..4d42c92d3 100644 --- a/packages/common-models/src/index.ts +++ b/packages/common-models/src/index.ts @@ -74,3 +74,4 @@ export * from "./login-provider"; export * from "./features"; export * from "./product-discussion"; export type { ScormContent } from "./scorm-content"; +export * from "./reply-token"; diff --git a/packages/common-models/src/reply-token.ts b/packages/common-models/src/reply-token.ts new file mode 100644 index 000000000..dbf6a61ff --- /dev/null +++ b/packages/common-models/src/reply-token.ts @@ -0,0 +1,14 @@ +export interface ReplyTokenPayload { + userId: string; + domainId: string; + entityId: string; + entityType: "community" | "product"; + commentId?: string; + parentReplyId?: string; +} + +export interface InboundEmailConfig { + provider: "ses" | "postmark" | "mailgun" | "sendgrid"; + endpoint?: string; + apiKey?: string; +}