Skip to content

inbounds and parsing #3

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 7 commits into from
Nov 11, 2025
Merged

inbounds and parsing #3

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
de336a8
setup api testing
marcrupt Nov 6, 2025
a7b2890
inbounds and parsing
marcrupt Nov 7, 2025
f6f05a9
threading the needle
marcrupt Nov 7, 2025
723c50b
tests/fixes
marcrupt Nov 8, 2025
4c62361
tests/fixes
marcrupt Nov 8, 2025
797d6bd
pr feedback
marcrupt Nov 10, 2025
a44ea0d
pr changes
marcrupt Nov 10, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 62 additions & 0 deletions api/bun.lock
View file
Open in desktop

Large diffs are not rendered by default.

3 changes: 0 additions & 3 deletions api/controllers/AuthController.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,12 +36,9 @@ export async function register(
name: "Default API Key",
});

const inboxEmail = await getNewRandomInboxEmail({ name: "inbox" });

const inbox = await createInbox({
organization_id: organization.id,
name: "My Inbox",
email: inboxEmail,
});

return { user, organization, apiKey, inbox };
Expand Down
4 changes: 2 additions & 2 deletions api/controllers/InboxController.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,15 +10,15 @@ export async function createInbox({
organization_id: string;
domain_id?: string;
name: string;
email: string;
email?: string;
}) {
const inbox = new Inbox();
inbox.organizationId = new mongoose.Types.ObjectId(organization_id);
inbox.domainId = domain_id
? new mongoose.Types.ObjectId(domain_id)
: undefined;
inbox.name = name;
inbox.email = email;
inbox.email = email || await getNewRandomInboxEmail({ name });
await inbox.save();
return inbox;
}
Expand Down
18 changes: 18 additions & 0 deletions api/controllers/MessageController.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,41 +1,59 @@
import mongoose from "mongoose";
import Message from "../db/mongo/schemas/Message";
import type { MessageStatus } from "../models/Message";

export async function createMessage({
organizationId,
inboxId,
threadId,
fromInboxId,
toInboxId,
from,
to,
externalMessageId,
subject,
text,
html,
status,
}: {
organizationId: string;
inboxId: string;
threadId: string;
fromInboxId?: string;
toInboxId?: string;
from: string;
to: string;
externalMessageId?: string;
subject: string;
text: string;
html: string;
status?: (typeof MessageStatus)[number];
}) {
const message = new Message();
message.organizationId = new mongoose.Types.ObjectId(organizationId);
message.inboxId = new mongoose.Types.ObjectId(inboxId);
message.threadId = new mongoose.Types.ObjectId(threadId);
message.fromInboxId = fromInboxId ? new mongoose.Types.ObjectId(fromInboxId) : undefined;
message.toInboxId = toInboxId ? new mongoose.Types.ObjectId(toInboxId) : undefined;
message.from = from;
message.to = to;
message.externalMessageId = externalMessageId;
message.subject = subject;
message.text = text;
message.html = html;
message.status = status;
await message.save();
return message;
}

export async function getMessageById(messageId: string) {
return await Message.findById(messageId);
}

export async function getMessagesByInboxId(inboxId: string) {
return await Message.find({ inboxId: new mongoose.Types.ObjectId(inboxId) });
}

export async function getMessageByExternalMessageId(externalMessageId: string) {
return await Message.findOne({ externalMessageId });
}
162 changes: 126 additions & 36 deletions api/controllers/SESController.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,10 +5,14 @@ import {
VerifyDomainDkimCommand,
} from "@aws-sdk/client-ses";
import type { SNSMessage } from "../models/SES";
import { getMessageById } from "./MessageController";
import { createMessage, getMessageByExternalMessageId, getMessageById } from "./MessageController";
import type { MessageStatus } from "../models/Message";
import type { WebhookEvents } from "../models/Webhook";
import { sendWebhookEvent } from "./WebhookAttemptController";
import { getInboxByEmail } from "./InboxController";
import { simpleParser } from "mailparser";
import EmailReplyParser from "email-reply-parser";
import { addMessageToThread, createThread } from "./ThreadController";

export const ses = new SESClient({
region: "us-east-2",
Expand Down Expand Up @@ -46,7 +50,6 @@ export async function sendSESMessage({
text: string;
html: string;
}) {
console.log("Sending SES message", { messageId, from, to, fromName, subject, text, html });
const command = new SendEmailCommand({
Source: fromName ? `${fromName} <${from}>` : from,
Destination: {
Expand Down Expand Up @@ -82,49 +85,136 @@ export async function sendSESMessage({
export async function handleDeliveryNotification(rawMessage: string) {
try {
const notification: SNSMessage = JSON.parse(rawMessage);
console.log("notification", notification);

const messageId = notification.mail?.tags?.["message"]?.[0];
if (!messageId) {
console.error(
"No sendook:message tag found in SES delivery notification"
);
return;
}
const message = await getMessageById(messageId);
if (!message) {
console.error("Message not found", messageId);
if (messageId) {
await handleOutboundSESMessage({
notification,
messageId,
});
return;
}

let status: (typeof MessageStatus)[number] | undefined;
let event: (typeof WebhookEvents)[number] | undefined;
if (notification.eventType === "Reject") {
status = "rejected";
event = "message.rejected";
} else if (notification.eventType === "Bounce") {
status = "bounced";
event = "message.bounced";
} else if (notification.eventType === "Complaint") {
status = "complained";
event = "message.complained";
} else if (notification.eventType === "Delivery") {
status = "delivered";
event = "message.delivered";
await handleInboundSESMessage({
notification,
});
} catch (error) {
console.error("Error handling SES delivery notification", error);
return;
}
}

export async function handleInboundSESMessage({
notification,
}: {
notification: SNSMessage;
}) {
if (!notification.mail.destination[0]) {
return;
}

const inbox = await getInboxByEmail(notification.mail.destination[0]);
if (!inbox) {
console.error("Inbox not found", notification.mail.destination[0]);
return;
}

const mail = await simpleParser(Buffer.from(notification.content, "base64").toString("utf-8"));
const content = new EmailReplyParser().read(mail.text ?? "");

const fromInboxId = await getInboxByEmail(notification.mail.source);

const reference = notification.mail.headers?.find(header => header.name === "References")?.value;
const replyToMessageId = reference?.match(/<([^@>]+)@us-east-2\.amazonses\.com>/)?.[1];

let threadId: string | undefined;
if (replyToMessageId) {
const message = await getMessageByExternalMessageId(replyToMessageId);
if (message) {
threadId = message.threadId.toString();
}
}

if (!threadId) {
const thread = await createThread({
organizationId: inbox.organizationId.toString(),
inboxId: inbox.id,
});
threadId = thread._id.toString();
}

const message = await createMessage({
organizationId: inbox.organizationId.toString(),
inboxId: inbox.id,
threadId,
from: notification.mail.source,
fromInboxId: fromInboxId?.id,
to: notification.mail.destination[0],
toInboxId: inbox.id,
subject: notification.mail.commonHeaders?.subject,
text: content.getVisibleText(),
html: content.getVisibleText(),
status: "received",
});
Comment on lines +147 to +159
Copy link

@coderabbitai coderabbitai bot Nov 7, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

HTML field should contain HTML content, not plain text.

Line 157 sets the html field to content.getVisibleText(), which returns plain text. This should either use the parsed HTML from the email or properly convert the text to HTML.

Consider this fix:

     text: content.getVisibleText(),
-    html: content.getVisibleText(),
+    html: mail.html || content.getVisibleText(),
     status: "received",

Committable suggestion skipped: line range outside the PR's diff.

🤖 Prompt for AI Agents 
In api/controllers/SESController.ts around lines 147 to 159, the message html
field is being set to plain text via content.getVisibleText(); change it to use
the parsed HTML from the email (e.g. content.getHtml() / content.getHTML() or
the parser method that returns HTML) and, if no HTML part exists, fall back to a
safe conversion of the visible text (escape HTML entities and replace newlines
with <br/> and wrap in a container) before assigning to html; ensure you use the
HTML-returning method when available and only convert text as a fallback.


await addMessageToThread({
threadId,
messageId: message._id.toString(),
});

await sendWebhookEvent({
organizationId: inbox.organizationId.toString(),
inboxId: inbox.id,
messageId: message.id,
event: "message.received",
payload: message,
});
}

export async function handleOutboundSESMessage({
notification,
messageId,
}: {
notification: SNSMessage;
messageId: string;
}) {
const message = await getMessageById(messageId);
if (!message) {
console.error("Message not found", messageId);
return;
}

let status: (typeof MessageStatus)[number] | undefined;
let event: (typeof WebhookEvents)[number] | undefined;
if (notification.eventType === "Reject") {
status = "rejected";
event = "message.rejected";
} else if (notification.eventType === "Bounce") {
status = "bounced";
event = "message.bounced";
} else if (notification.eventType === "Complaint") {
status = "complained";
event = "message.complained";
} else if (notification.eventType === "Delivery") {
status = "delivered";
event = "message.delivered";
}

if (status) {
message.status = status;
await message.save();
}

if (event) {
await sendWebhookEvent({
organizationId: message.organizationId.toString(),
inboxId: message.inboxId.toString(),
messageId: message.id,
event,
payload: message,
});
}
} catch (error) {
console.error("Error handling SES delivery notification", error);
if (!event) {
return;
}

await sendWebhookEvent({
organizationId: message.organizationId.toString(),
inboxId: message.inboxId.toString(),
messageId: message.id,
event,
payload: message,
});
}
43 changes: 43 additions & 0 deletions api/controllers/ThreadController.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,43 @@
import mongoose from "mongoose";
import Thread from "../db/mongo/schemas/Thread";

export async function createThread({
organizationId,
inboxId,
}: {
organizationId: string;
inboxId: string;
}) {
const thread = new Thread();
thread.organizationId = new mongoose.Types.ObjectId(organizationId);
thread.inboxId = new mongoose.Types.ObjectId(inboxId);
thread.messages = new mongoose.Types.Array<mongoose.Types.ObjectId>();
await thread.save();
return thread;
}
Comment on lines +4 to +17
Copy link

@coderabbitai coderabbitai bot Nov 8, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Add error handling and input validation.

This function lacks try-catch error handling and input validation. Invalid ObjectId strings will throw unhandled exceptions that could crash the service or leak error details.

Wrap in try-catch and validate inputs:

 export async function createThread({
   organizationId,
   inboxId,
 }: {
   organizationId: string;
   inboxId: string;
 }) {
+  try {
+    // Validate ObjectId format
+    if (!mongoose.Types.ObjectId.isValid(organizationId) || !mongoose.Types.ObjectId.isValid(inboxId)) {
+      throw new Error("Invalid organizationId or inboxId format");
+    }
+
     const thread = new Thread();
     thread.organizationId = new mongoose.Types.ObjectId(organizationId);
     thread.inboxId = new mongoose.Types.ObjectId(inboxId);
     thread.messages = new mongoose.Types.Array<mongoose.Types.ObjectId>();
     await thread.save();
     return thread;
+  } catch (error) {
+    console.error("Error creating thread:", error);
+    throw error;
+  }
 }
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
export async function createThread({
organizationId,
inboxId,
}: {
organizationId: string;
inboxId: string;
}) {
const thread = new Thread();
thread.organizationId = new mongoose.Types.ObjectId(organizationId);
thread.inboxId = new mongoose.Types.ObjectId(inboxId);
thread.messages = new mongoose.Types.Array<mongoose.Types.ObjectId>();
await thread.save();
return thread;
}
export async function createThread({
organizationId,
inboxId,
}: {
organizationId: string;
inboxId: string;
}) {
try {
// Validate ObjectId format
if (!mongoose.Types.ObjectId.isValid(organizationId) || !mongoose.Types.ObjectId.isValid(inboxId)) {
throw new Error("Invalid organizationId or inboxId format");
}
const thread = new Thread();
thread.organizationId = new mongoose.Types.ObjectId(organizationId);
thread.inboxId = new mongoose.Types.ObjectId(inboxId);
thread.messages = new mongoose.Types.Array<mongoose.Types.ObjectId>();
await thread.save();
return thread;
} catch (error) {
console.error("Error creating thread:", error);
throw error;
}
}
🤖 Prompt for AI Agents 
In api/controllers/ThreadController.ts around lines 4–17, validate that
organizationId and inboxId are non-empty strings and pass
mongoose.Types.ObjectId.isValid before constructing ObjectId instances; if
validation fails, throw a controlled error (e.g., BadRequest) with a safe
message. Wrap the creation and thread.save() in a try-catch, log a concise error
message (without leaking stack or raw input) and rethrow or return a controlled
error response so callers can handle it; ensure thread.messages is initialized
as a plain array of ObjectId placeholders and only create ObjectId instances
after validation.


export async function addMessageToThread({
threadId,
messageId,
}: {
threadId: string;
messageId: string;
}) {
const thread = await Thread.findById(threadId);
if (!thread) {
return null;
}
thread.messages.push(new mongoose.Types.ObjectId(messageId));
await thread.save();
return thread;
}
Comment on lines +19 to +33
Copy link

@coderabbitai coderabbitai bot Nov 8, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Add error handling, input validation, and duplicate check.

This function has several issues:

  1. No try-catch error handling for invalid ObjectIds or database errors
  2. No validation of input ID formats
  3. No check for duplicate messages in the thread
  4. Potential race condition with concurrent addMessageToThread calls

Apply these improvements:

 export async function addMessageToThread({
   threadId,
   messageId,
 }: {
   threadId: string;
   messageId: string;
 }) {
+  try {
+    if (!mongoose.Types.ObjectId.isValid(threadId) || !mongoose.Types.ObjectId.isValid(messageId)) {
+      throw new Error("Invalid threadId or messageId format");
+    }
+
     const thread = await Thread.findById(threadId);
     if (!thread) {
       return null;
     }
+    
+    // Check for duplicate
+    const messageObjectId = new mongoose.Types.ObjectId(messageId);
+    if (thread.messages.some(id => id.equals(messageObjectId))) {
+      return thread; // Already exists
+    }
+    
-    thread.messages.push(new mongoose.Types.ObjectId(messageId));
+    thread.messages.push(messageObjectId);
     await thread.save();
     return thread;
+  } catch (error) {
+    console.error("Error adding message to thread:", error);
+    throw error;
+  }
 }


export async function getThreadsByInboxId(inboxId: string) {
const threads = await Thread.find({ inboxId: new mongoose.Types.ObjectId(inboxId) });
return threads;
}
Comment on lines +35 to +38
Copy link

@coderabbitai coderabbitai bot Nov 8, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Add error handling and consider pagination.

This function lacks error handling and could return large result sets without pagination.

Add error handling:

 export async function getThreadsByInboxId(inboxId: string) {
+  try {
+    if (!mongoose.Types.ObjectId.isValid(inboxId)) {
+      throw new Error("Invalid inboxId format");
+    }
+
     const threads = await Thread.find({ inboxId: new mongoose.Types.ObjectId(inboxId) });
     return threads;
+  } catch (error) {
+    console.error("Error fetching threads:", error);
+    throw error;
+  }
 }

Consider adding pagination parameters (limit, skip) to handle inboxes with many threads.

🤖 Prompt for AI Agents 
In api/controllers/ThreadController.ts around lines 35 to 38, the function has
no error handling and can return unbounded results; wrap the DB call in a
try/catch that validates inboxId and rethrows or returns a controlled error, and
add pagination parameters (e.g., limit and skip or page and size) with sane
defaults and a maximum cap to prevent huge responses; apply
.limit(limit).skip(skip) to the query and optionally return metadata (total
count, page) or throw a 400 for invalid input and a 500 for DB errors.


export async function getThreadById(threadId: string) {
const thread = await Thread.findById(threadId).populate("messages");
return thread;
}
Comment on lines +40 to +43
Copy link

@coderabbitai coderabbitai bot Nov 8, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Add error handling and input validation.

Missing error handling and input validation could cause unhandled exceptions.

 export async function getThreadById(threadId: string) {
+  try {
+    if (!mongoose.Types.ObjectId.isValid(threadId)) {
+      throw new Error("Invalid threadId format");
+    }
+
     const thread = await Thread.findById(threadId).populate("messages");
     return thread;
+  } catch (error) {
+    console.error("Error fetching thread:", error);
+    throw error;
+  }
 }
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
export async function getThreadById(threadId: string) {
const thread = await Thread.findById(threadId).populate("messages");
return thread;
}
export async function getThreadById(threadId: string) {
try {
if (!mongoose.Types.ObjectId.isValid(threadId)) {
throw new Error("Invalid threadId format");
}
const thread = await Thread.findById(threadId).populate("messages");
return thread;
} catch (error) {
console.error("Error fetching thread:", error);
throw error;
}
}
🤖 Prompt for AI Agents 
In api/controllers/ThreadController.ts around lines 40 to 43, the function lacks
input validation and error handling; validate the threadId (e.g., ensure it's
present and a valid Mongo ObjectId using mongoose.Types.ObjectId.isValid) before
calling Thread.findById, and wrap the database call in a try/catch to handle and
log DB errors and return a controlled response (null or throw a descriptive
error) instead of letting exceptions bubble up.

14 changes: 9 additions & 5 deletions api/db/mongo/schemas/Message.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,11 +14,11 @@ const messageSchema = new mongoose.Schema(
ref: "Inbox",
required: true,
},
// threadId: {
// type: mongoose.Schema.Types.ObjectId,
// ref: "Thread",
// required: true,
// },
threadId: {
type: mongoose.Schema.Types.ObjectId,
ref: "Thread",
required: true,
},
fromInboxId: {
type: mongoose.Schema.Types.ObjectId,
ref: "Inbox",
Expand All @@ -33,6 +33,10 @@ const messageSchema = new mongoose.Schema(
ref: "Inbox",
required: false,
},
externalMessageId: {
type: String,
required: false,
},
to: {
type: String,
required: true,
Expand Down
10 changes: 3 additions & 7 deletions api/db/mongo/schemas/Thread.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,12 +1,8 @@
import mongoose from "mongoose";
import type IMessage from "../../../models/Message";
import type IThread from "../../../models/Thread";

const messageSchema = new mongoose.Schema(
const threadSchema = new mongoose.Schema(
{
name: {
type: String,
required: true,
},
organizationId: {
type: mongoose.Schema.Types.ObjectId,
ref: "Organization",
Expand All @@ -28,4 +24,4 @@ const messageSchema = new mongoose.Schema(
}
);

export default mongoose.model<IMessage>("Message", messageSchema);
export default mongoose.model<IThread>("Thread", threadSchema);
4 changes: 2 additions & 2 deletions api/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@ import cors from "cors";
import authRouter from "./routes/auth";
import organizationsRouter from "./routes/organizations";
import webhooksRouter from "./routes/webhooks";
import inboxesRouter from "./routes/inboxes";
import v1Router from "./routes/v1/index";
Copy link

@coderabbitai coderabbitai bot Nov 7, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Breaking API change: Document the migration path.

Replacing the /inboxes endpoint with /v1 is a breaking change that will affect existing API consumers. Consider maintaining both endpoints during a transition period or providing clear migration documentation.

Would you like me to generate a migration guide or help implement backward compatibility with a deprecation notice?

🤖 Prompt for AI Agents 
In api/index.ts around line 10, the import indicates the router was switched to
/v1 which is a breaking change for existing /inboxes consumers; restore backward
compatibility by mounting the existing v1Router under both routes during a
transition (mount /inboxes to delegate to the same router or a thin proxy that
emits a deprecation response header/body), add a server-side deprecation warning
(e.g., Deprecation and Sunset headers or a JSON notice) for /inboxes, and update
API docs and changelog to include a clear migration path and timeline for
removing /inboxes.


startMongo();

Expand Down Expand Up @@ -47,8 +47,8 @@ app.get("/health", (req, res) => {

app.use("/auth", authRouter);
app.use("/organizations", organizationsRouter);
app.use("/inboxes", inboxesRouter);
app.use("/webhooks", webhooksRouter);
app.use("/v1", v1Router);

app.listen(port, () => {
console.log(`Listening on port ${port}...`);
Expand Down
Loading