diff --git a/app/app.tsx b/app/app.tsx index bb28d31..2ebed28 100644 --- a/app/app.tsx +++ b/app/app.tsx @@ -9,9 +9,9 @@ import type { CurrentUser } from "@/features/auth/types"; import { InboxPage } from "@/features/inbox/inbox-page"; import { listMailboxes } from "@/features/mailboxes/api"; import type { Mailbox } from "@/features/mailboxes/types"; -import { listMessages } from "@/features/messages/api"; +import { listConversations, listMessages } from "@/features/messages/api"; import { mergeIncomingMessageIds } from "@/features/messages/incoming-sound"; -import type { MessageSummary } from "@/features/messages/types"; +import type { ConversationSummary } from "@/features/messages/types"; import { SettingsPage } from "@/features/settings/settings-page"; import { getSetupStatus } from "@/features/setup/api"; import { SetupPage } from "@/features/setup/setup-page"; @@ -33,7 +33,7 @@ export function App(): React.ReactElement { const [user, setUser] = React.useState(null); const [mailboxes, setMailboxes] = React.useState([]); const [users, setUsers] = React.useState([]); - const [messages, setMessages] = React.useState([]); + const [conversations, setConversations] = React.useState([]); const [mailboxId, setMailboxId] = React.useState("all"); const [search, setSearch] = React.useState(""); const [composeOpen, setComposeOpen] = React.useState(false); @@ -88,19 +88,14 @@ export function App(): React.ReactElement { } }, [loadWorkspace]); - const reloadMessages = React.useCallback(async () => { + const reloadConversations = React.useCallback(async () => { if (!user || activeFolder === "settings") return; - const folder = activeFolder === "starred" ? undefined : activeFolder; - const nextMessages = await listMessages({ - folder, + const nextConversations = await listConversations({ + folder: activeFolder, mailboxId: mailboxId === "all" ? undefined : mailboxId, search: search || undefined }); - const filtered = - activeFolder === "starred" - ? nextMessages.filter((message) => message.starredAt) - : nextMessages; - setMessages(filtered); + setConversations(nextConversations); }, [activeFolder, mailboxId, search, user]); React.useEffect(() => { @@ -108,10 +103,10 @@ export function App(): React.ReactElement { }, [reload]); React.useEffect(() => { - void reloadMessages().catch((error) => { - toast.error(error instanceof Error ? error.message : "Messages failed to load."); + void reloadConversations().catch((error) => { + toast.error(error instanceof Error ? error.message : "Conversations failed to load."); }); - }, [reloadMessages]); + }, [reloadConversations]); React.useEffect(() => { if (!user || (user.role !== "owner" && user.role !== "admin")) return; @@ -164,13 +159,13 @@ export function App(): React.ReactElement { if (!user || activeFolder === "settings") return; const interval = window.setInterval(() => { - void reloadMessages().catch(() => { + void reloadConversations().catch(() => { // Keep background refresh failures quiet; the next interval will retry. }); }, 10_000); return () => window.clearInterval(interval); - }, [activeFolder, reloadMessages, user]); + }, [activeFolder, reloadConversations, user]); React.useEffect(() => { if (!user || isLoading || route.kind !== "settings") return; @@ -230,7 +225,7 @@ export function App(): React.ReactElement { onSearchChange={setSearch} onSignedOut={() => { setUser(null); - setMessages([]); + setConversations([]); }} >
@@ -249,10 +244,10 @@ export function App(): React.ReactElement { ) : ( void reloadMessages()} + onRefresh={() => void reloadConversations()} onMessageRouteChange={(folder, messageId) => navigate({ kind: "mail", folder, messageId }) } @@ -271,7 +266,7 @@ export function App(): React.ReactElement { mode="new" open={composeOpen} onOpenChange={setComposeOpen} - onSent={() => void reloadMessages()} + onSent={() => void reloadConversations()} /> ) : null} diff --git a/app/features/inbox/inbox-page.tsx b/app/features/inbox/inbox-page.tsx index 7c47c8b..3950b87 100644 --- a/app/features/inbox/inbox-page.tsx +++ b/app/features/inbox/inbox-page.tsx @@ -1,18 +1,21 @@ import * as React from "react"; import type { Mailbox } from "@/features/mailboxes/types"; -import { getMessageThread, runMessageAction } from "@/features/messages/api"; +import { getMessageThread, runConversationAction } from "@/features/messages/api"; import { MessageDetail } from "@/features/messages/message-detail"; import { MessageList } from "@/features/messages/message-list"; -import type { MessageDetail as MessageDetailType, MessageSummary } from "@/features/messages/types"; +import type { + ConversationSummary, + MessageDetail as MessageDetailType +} from "@/features/messages/types"; import { cn } from "@/lib/cn"; import type { MailFolderId } from "@/lib/routes"; import { mailFolders } from "@/lib/routes"; type InboxPageProps = { activeFolder: MailFolderId; + conversations: ConversationSummary[]; mailboxes: Mailbox[]; - messages: MessageSummary[]; selectedId: string | null; onRefresh: () => void; onMessageRouteChange: (folder: MailFolderId, messageId: string | null) => void; @@ -21,8 +24,8 @@ type InboxPageProps = { export function InboxPage({ activeFolder, + conversations, mailboxes, - messages, selectedId, onRefresh, onMessageRouteChange, @@ -57,16 +60,21 @@ export function InboxPage({ .then((messages) => { if (cancelled) return; setThread(messages); - const selected = messages.find((message) => message.id === selectedId); - if (selected?.readAt === null) { - void runMessageAction(selectedId, "read") + if ( + messages.some((message) => message.direction === "inbound" && message.readAt === null) + ) { + void runConversationAction(selectedId, "read", activeFolder) .then((updated) => { if (cancelled) return; - setThread((current) => - current.map((message) => - message.id === updated.id ? { ...message, readAt: updated.readAt } : message - ) - ); + if (updated.affected > 0) { + setThread((current) => + current.map((message) => + message.direction === "inbound" + ? { ...message, readAt: new Date().toISOString() } + : message + ) + ); + } onRefreshRef.current(); }) .catch(() => undefined); @@ -83,20 +91,41 @@ export function InboxPage({ return () => { cancelled = true; }; - }, [selectedId]); + }, [activeFolder, selectedId]); - async function handleAction(action: Parameters[1]) { + const selectedThreadId = + thread[0]?.threadId ?? + conversations.find((conversation) => conversation.id === selectedId)?.threadId ?? + null; + const selectedConversation = conversations.find( + (conversation) => conversation.threadId === selectedThreadId + ); + const readerSelectedId = selectedConversation?.id ?? selectedId; + + React.useEffect(() => { + if ( + !selectedId || + !selectedConversation || + thread.some((message) => message.id === selectedConversation.id) + ) { + return; + } + void loadThread(selectedConversation.id); + }, [loadThread, selectedConversation, selectedId, thread]); + + async function handleAction(action: Parameters[1]) { if (!selectedId) return; - const updated = await runMessageAction(selectedId, action); + await runConversationAction(selectedId, action, activeFolder); onRefresh(); - await loadThread(selectedId); if ( action === "archive" || action === "trash" || (activeFolder === "starred" && action === "unstar") ) { - onMessageRouteChange(updated.folder, updated.id); + onMessageRouteChange(activeFolder, null); + return; } + await loadThread(selectedId); } return ( @@ -111,15 +140,17 @@ export function InboxPage({

{activeLabel} - Messages + Conversations

- {messages.length} + + {conversations.length} +
onSelect(message.id)} + conversations={conversations} + selectedThreadId={selectedThreadId} + onSelect={(conversation) => onSelect(conversation.id)} />
void handleAction(action)} onBack={() => onMessageRouteChange(activeFolder, null)} onSent={() => { diff --git a/app/features/messages/api.ts b/app/features/messages/api.ts index c0f4910..eccc42d 100644 --- a/app/features/messages/api.ts +++ b/app/features/messages/api.ts @@ -1,5 +1,6 @@ import { apiGet, apiPost } from "@/lib/api-client"; -import type { MessageDetail, MessageHtml, MessageSummary } from "./types"; +import type { MailFolderId } from "@/lib/routes"; +import type { ConversationSummary, MessageDetail, MessageHtml, MessageSummary } from "./types"; export type MessageListParams = { folder?: string | undefined; @@ -16,6 +17,15 @@ export async function listMessages(params: MessageListParams): Promise(`/api/messages${suffix}`); } +export async function listConversations( + params: MessageListParams & { folder: MailFolderId } +): Promise { + const query = new URLSearchParams({ folder: params.folder }); + if (params.mailboxId) query.set("mailboxId", params.mailboxId); + if (params.search) query.set("search", params.search); + return apiGet(`/api/conversations?${query.toString()}`); +} + export async function getMessage(id: string): Promise { return apiGet(`/api/messages/${id}`); } @@ -39,3 +49,11 @@ export async function runMessageAction( ): Promise { return apiPost(`/api/messages/${id}/${action}`); } + +export async function runConversationAction( + id: string, + action: "read" | "unread" | "star" | "unstar" | "archive" | "trash", + folder: MailFolderId +): Promise<{ affected: number; threadId: string }> { + return apiPost(`/api/conversations/${id}/${action}`, { folder }); +} diff --git a/app/features/messages/message-detail.tsx b/app/features/messages/message-detail.tsx index 776c12f..9fe2a22 100644 --- a/app/features/messages/message-detail.tsx +++ b/app/features/messages/message-detail.tsx @@ -54,6 +54,10 @@ export function MessageDetail({ ? selected : ([...messages].reverse().find((message) => message.direction === "inbound") ?? selected); const composeTarget = composeMode === "reply" ? replyTarget : selected; + const isUnread = messages.some( + (message) => message.direction === "inbound" && message.readAt === null + ); + const isStarred = messages.some((message) => message.starredAt !== null); return (
@@ -74,21 +78,21 @@ export function MessageDetail({
onAction(selected.readAt ? "unread" : "read")} + label={isUnread ? "Mark conversation read" : "Mark conversation unread"} + onClick={() => onAction(isUnread ? "read" : "unread")} > onAction(selected.starredAt ? "unstar" : "star")} + label={isStarred ? "Unstar conversation" : "Star conversation"} + onClick={() => onAction(isStarred ? "unstar" : "star")} > - onAction("archive")}> + onAction("archive")}> - onAction("trash")}> + onAction("trash")}>
diff --git a/app/features/messages/message-list-item.tsx b/app/features/messages/message-list-item.tsx index a3b3235..442a7ae 100644 --- a/app/features/messages/message-list-item.tsx +++ b/app/features/messages/message-list-item.tsx @@ -1,29 +1,38 @@ -import { Paperclip, Star } from "lucide-react"; +import { MessagesSquare, Paperclip, Star } from "lucide-react"; import type * as React from "react"; import { Badge } from "@/components/ui/badge"; import { cn } from "@/lib/cn"; import { formatDateTime } from "@/lib/format"; -import type { MessageSummary } from "./types"; +import type { MailFolderId } from "@/lib/routes"; +import type { ConversationSummary } from "./types"; type MessageListItemProps = { + activeFolder: MailFolderId; + conversation: ConversationSummary; href: string; - message: MessageSummary; isActive: boolean; - onSelect: (message: MessageSummary) => void; + onSelect: (conversation: ConversationSummary) => void; }; export function MessageListItem({ + activeFolder, + conversation, href, - message, isActive, onSelect }: MessageListItemProps): React.ReactElement { + const isUnread = conversation.unreadCount > 0; + const correspondent = + conversation.direction === "inbound" + ? conversation.fromAddress + : `To: ${conversation.to[0] ?? "recipient"}`; + return ( { @@ -37,12 +46,12 @@ export function MessageListItem({ return; } event.preventDefault(); - onSelect(message); + onSelect(conversation); }} >
- {message.readAt === null && ( + {isUnread && ( )} - - {message.fromAddress} + + {correspondent}
- {formatDateTime(message.receivedAt ?? message.sentAt ?? message.createdAt)} + {formatDateTime(conversation.receivedAt ?? conversation.sentAt ?? conversation.createdAt)}
- {message.starredAt && } - - {message.subject} + {conversation.isStarred && } + + {conversation.subject} - {message.hasAttachments && } + {conversation.messageCount > 1 && ( + + + {conversation.messageCount} messages + + )} + {conversation.hasAttachments && }

- {message.snippet || "No preview"} + {conversation.snippet || "No preview"}

- {message.folder === "catchall" && ( + {activeFolder === "catchall" && ( Unknown @@ -83,9 +101,9 @@ export function EmptyMessageList(): React.ReactElement { return (
- +
-
No messages in this view
+
No conversations in this view
); } diff --git a/app/features/messages/message-list.tsx b/app/features/messages/message-list.tsx index b070442..346837d 100644 --- a/app/features/messages/message-list.tsx +++ b/app/features/messages/message-list.tsx @@ -2,33 +2,34 @@ import type * as React from "react"; import { appRoutePath, type MailFolderId } from "@/lib/routes"; import { EmptyMessageList, MessageListItem } from "./message-list-item"; -import type { MessageSummary } from "./types"; +import type { ConversationSummary } from "./types"; type MessageListProps = { activeFolder: MailFolderId; - messages: MessageSummary[]; - selectedId: string | null; - onSelect: (message: MessageSummary) => void; + conversations: ConversationSummary[]; + selectedThreadId: string | null; + onSelect: (conversation: ConversationSummary) => void; }; export function MessageList({ activeFolder, - messages, - selectedId, + conversations, + selectedThreadId, onSelect }: MessageListProps): React.ReactElement { - if (messages.length === 0) { + if (conversations.length === 0) { return ; } return (
- {messages.map((message) => ( + {conversations.map((conversation) => ( ))} diff --git a/app/features/messages/types.ts b/app/features/messages/types.ts index 21d7342..aec44e8 100644 --- a/app/features/messages/types.ts +++ b/app/features/messages/types.ts @@ -18,6 +18,12 @@ export type MessageSummary = { createdAt: string; }; +export type ConversationSummary = MessageSummary & { + isStarred: boolean; + messageCount: number; + unreadCount: number; +}; + export type MessageDetail = MessageSummary & { cc: string[]; bcc: string[]; diff --git a/migrations/0005_rebuild_threads.sql b/migrations/0005_rebuild_threads.sql new file mode 100644 index 0000000..e570e04 --- /dev/null +++ b/migrations/0005_rebuild_threads.sql @@ -0,0 +1,103 @@ +CREATE TABLE thread_rebuild_map ( + message_id TEXT PRIMARY KEY NOT NULL, + root_message_id TEXT NOT NULL +); + +WITH RECURSIVE parent_links(child_id, parent_id) AS ( + SELECT child.id, + COALESCE( + ( + SELECT parent.id + FROM messages parent + WHERE child.in_reply_to IS NOT NULL + AND parent.id <> child.id + AND parent.mailbox_id IS child.mailbox_id + AND (parent.message_id = child.in_reply_to OR parent.id = child.in_reply_to) + ORDER BY + COALESCE(parent.received_at, parent.sent_at, parent.created_at) DESC + LIMIT 1 + ), + ( + SELECT parent.id + FROM messages parent + WHERE child.in_reply_to IS NOT NULL + AND parent.id <> child.id + AND (parent.message_id = child.in_reply_to OR parent.id = child.in_reply_to) + ORDER BY + COALESCE(parent.received_at, parent.sent_at, parent.created_at) DESC + LIMIT 1 + ), + ( + SELECT parent.id + FROM json_each(child.references_json) reference + JOIN messages parent + ON parent.message_id = reference.value OR parent.id = reference.value + WHERE parent.id <> child.id + AND parent.mailbox_id IS child.mailbox_id + ORDER BY + CAST(reference.key AS INTEGER) DESC, + COALESCE(parent.received_at, parent.sent_at, parent.created_at) DESC + LIMIT 1 + ), + ( + SELECT parent.id + FROM json_each(child.references_json) reference + JOIN messages parent + ON parent.message_id = reference.value OR parent.id = reference.value + WHERE parent.id <> child.id + ORDER BY + CAST(reference.key AS INTEGER) DESC, + COALESCE(parent.received_at, parent.sent_at, parent.created_at) DESC + LIMIT 1 + ) + ) + FROM messages child +), +walked(message_id, root_message_id, depth) AS ( + SELECT child_id, child_id, 0 + FROM parent_links + WHERE parent_id IS NULL + + UNION ALL + + SELECT parent_links.child_id, walked.root_message_id, walked.depth + 1 + FROM walked + JOIN parent_links ON parent_links.parent_id = walked.message_id + WHERE walked.depth < 100 +) +INSERT INTO thread_rebuild_map (message_id, root_message_id) +SELECT message_id, root_message_id +FROM walked; + +INSERT OR IGNORE INTO thread_rebuild_map (message_id, root_message_id) +SELECT id, id +FROM messages; + +INSERT INTO threads (id, subject_normalized, last_message_at, created_at, updated_at) +SELECT + 'thr_rebuilt_' || mapping.root_message_id, + root_thread.subject_normalized, + MAX(COALESCE(message.received_at, message.sent_at, message.created_at)), + MIN(message.created_at), + MAX(message.updated_at) +FROM thread_rebuild_map mapping +JOIN messages message ON message.id = mapping.message_id +JOIN messages root_message ON root_message.id = mapping.root_message_id +JOIN threads root_thread ON root_thread.id = root_message.thread_id +GROUP BY mapping.root_message_id, root_thread.subject_normalized; + +UPDATE messages +SET thread_id = 'thr_rebuilt_' || ( + SELECT mapping.root_message_id + FROM thread_rebuild_map mapping + WHERE mapping.message_id = messages.id +); + +DELETE FROM threads +WHERE NOT EXISTS ( + SELECT 1 + FROM messages + WHERE messages.thread_id = threads.id +); + +DROP TABLE thread_rebuild_map; diff --git a/test/integration/worker/auth.test.ts b/test/integration/worker/auth.test.ts index 6ea5e72..c67e624 100644 --- a/test/integration/worker/auth.test.ts +++ b/test/integration/worker/auth.test.ts @@ -5,6 +5,7 @@ import initialMigration from "../../../migrations/0001_initial.sql?raw"; import workspaceMigration from "../../../migrations/0002_workspace.sql?raw"; import oauthResourcesMigration from "../../../migrations/0003_oauth_resources.sql?raw"; import conversationMigration from "../../../migrations/0004_conversations.sql?raw"; +import threadRebuildMigration from "../../../migrations/0005_rebuild_threads.sql?raw"; import { createAuth } from "../../../worker/auth/auth"; const origin = "https://hqbase.test"; @@ -30,6 +31,7 @@ describe("Better Auth schema", () => { await applyMigration(oauthResourcesMigration); await applyMigration(conversationMigration); + await applyMigration(threadRebuildMigration); }); it("backfills the Better Auth 1.7 account identity without losing credential rows", async () => { diff --git a/test/integration/worker/conversations.test.ts b/test/integration/worker/conversations.test.ts new file mode 100644 index 0000000..7d4332c --- /dev/null +++ b/test/integration/worker/conversations.test.ts @@ -0,0 +1,189 @@ +import { env } from "cloudflare:test"; +import { beforeAll, describe, expect, it } from "vitest"; + +import initialMigration from "../../../migrations/0001_initial.sql?raw"; +import workspaceMigration from "../../../migrations/0002_workspace.sql?raw"; +import oauthResourcesMigration from "../../../migrations/0003_oauth_resources.sql?raw"; +import conversationMigration from "../../../migrations/0004_conversations.sql?raw"; +import threadRebuildMigration from "../../../migrations/0005_rebuild_threads.sql?raw"; +import { + listConversations, + updateConversationAction +} from "../../../worker/features/messages/conversation-queries"; + +describe("conversation persistence", () => { + beforeAll(async () => { + for (const migration of [ + initialMigration, + workspaceMigration, + oauthResourcesMigration, + conversationMigration + ]) { + await applyMigration(migration); + } + await env.DB.prepare( + `INSERT INTO mailboxes (id, address, display_name, is_active, created_at, updated_at) + VALUES ('mbx_conversations', 'support@example.com', 'Support', 1, ?, ?)` + ) + .bind("2026-07-28T12:00:00.000Z", "2026-07-28T12:00:00.000Z") + .run(); + await env.DB.prepare( + `INSERT INTO threads (id, subject_normalized, last_message_at, created_at, updated_at) + VALUES ('thr_subject', 'status update', ?, ?, ?)` + ) + .bind("2026-07-28T14:00:00.000Z", "2026-07-28T12:00:00.000Z", "2026-07-28T14:00:00.000Z") + .run(); + await insertLegacyMessage({ + id: "msg_root_a", + direction: "inbound", + folder: "inbox", + from: "alice@example.com", + messageId: "", + occurredAt: "2026-07-28T12:00:00.000Z" + }); + await insertLegacyMessage({ + id: "msg_reply_a", + direction: "outbound", + folder: "sent", + from: "support@example.com", + inReplyTo: "", + messageId: "", + occurredAt: "2026-07-28T13:00:00.000Z", + references: [""] + }); + await insertLegacyMessage({ + id: "msg_root_b", + direction: "inbound", + folder: "inbox", + from: "bob@example.com", + messageId: "", + occurredAt: "2026-07-28T14:00:00.000Z" + }); + await applyMigration(threadRebuildMigration); + }); + + it("repairs subject-only history from message headers", async () => { + const rows = await env.DB.prepare("SELECT id, thread_id FROM messages ORDER BY id").all<{ + id: string; + thread_id: string; + }>(); + const threadByMessage = new Map(rows.results.map((row) => [row.id, row.thread_id])); + + expect(threadByMessage.get("msg_reply_a")).toBe(threadByMessage.get("msg_root_a")); + expect(threadByMessage.get("msg_root_b")).not.toBe(threadByMessage.get("msg_root_a")); + }); + + it("lists one latest-message row per conversation and applies scoped actions", async () => { + const filters = { folder: "inbox" as const, mailboxIds: ["mbx_conversations"] }; + const initial = await listConversations(env.DB, filters); + const alice = initial.find((conversation) => conversation.messageCount === 2); + + expect(initial).toHaveLength(2); + expect(alice).toMatchObject({ + direction: "outbound", + id: "msg_reply_a", + isStarred: false, + unreadCount: 1 + }); + + await expect( + updateConversationAction(env.DB, { + action: "read", + activeFolder: "inbox", + mailboxIds: ["mbx_conversations"], + messageId: "msg_root_a" + }) + ).resolves.toMatchObject({ affected: 1 }); + await expect( + updateConversationAction(env.DB, { + action: "star", + activeFolder: "inbox", + mailboxIds: ["mbx_conversations"], + messageId: "msg_root_a" + }) + ).resolves.toMatchObject({ affected: 2 }); + + const starred = await listConversations(env.DB, { + folder: "starred", + mailboxIds: ["mbx_conversations"] + }); + expect(starred).toHaveLength(1); + expect(starred[0]).toMatchObject({ id: "msg_reply_a", isStarred: true, unreadCount: 0 }); + + await expect( + updateConversationAction(env.DB, { + action: "archive", + activeFolder: "inbox", + mailboxIds: ["mbx_conversations"], + messageId: "msg_root_a" + }) + ).resolves.toMatchObject({ affected: 1 }); + expect(await listConversations(env.DB, filters)).toHaveLength(1); + expect( + await listConversations(env.DB, { + folder: "sent", + mailboxIds: ["mbx_conversations"] + }) + ).toHaveLength(1); + + await expect( + updateConversationAction(env.DB, { + action: "trash", + activeFolder: "sent", + mailboxIds: ["mbx_conversations"], + messageId: "msg_root_a" + }) + ).resolves.toMatchObject({ affected: 1 }); + }); +}); + +async function insertLegacyMessage(input: { + direction: "inbound" | "outbound"; + folder: "inbox" | "sent"; + from: string; + id: string; + inReplyTo?: string; + messageId: string; + occurredAt: string; + references?: string[]; +}): Promise { + const inbound = input.direction === "inbound"; + await env.DB.prepare( + `INSERT INTO messages ( + id, thread_id, mailbox_id, direction, folder, from_address, to_json, cc_json, bcc_json, + subject, snippet, text_body, message_id, dedupe_key, in_reply_to, references_json, + received_at, sent_at, read_at, has_attachments, created_at, updated_at + ) VALUES ( + ?, 'thr_subject', 'mbx_conversations', ?, ?, ?, ?, '[]', '[]', + 'Status update', ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ? + )` + ) + .bind( + input.id, + input.direction, + input.folder, + input.from, + JSON.stringify([inbound ? "support@example.com" : "alice@example.com"]), + input.id, + input.id, + input.messageId, + `${input.id}:support@example.com`, + input.inReplyTo ?? null, + JSON.stringify(input.references ?? []), + inbound ? input.occurredAt : null, + inbound ? null : input.occurredAt, + inbound ? null : input.occurredAt, + input.occurredAt, + input.occurredAt + ) + .run(); +} + +async function applyMigration(source: string): Promise { + for (const statement of source + .split(";") + .map((value) => value.trim()) + .filter(Boolean)) { + await env.DB.prepare(statement).run(); + } +} diff --git a/test/integration/worker/mcp.test.ts b/test/integration/worker/mcp.test.ts index ee500ea..9aafaab 100644 --- a/test/integration/worker/mcp.test.ts +++ b/test/integration/worker/mcp.test.ts @@ -5,6 +5,7 @@ import initialMigration from "../../../migrations/0001_initial.sql?raw"; import workspaceMigration from "../../../migrations/0002_workspace.sql?raw"; import oauthResourcesMigration from "../../../migrations/0003_oauth_resources.sql?raw"; import conversationMigration from "../../../migrations/0004_conversations.sql?raw"; +import threadRebuildMigration from "../../../migrations/0005_rebuild_threads.sql?raw"; import { hashOAuthToken } from "../../../worker/auth/oauth-token"; const origin = "https://hqbase.test"; @@ -18,7 +19,8 @@ describe("HQBase MCP server", () => { initialMigration, workspaceMigration, oauthResourcesMigration, - conversationMigration + conversationMigration, + threadRebuildMigration ]) { await applyMigration(migration); } diff --git a/test/unit/app/layout/mail-shell.test.tsx b/test/unit/app/layout/mail-shell.test.tsx index 64fcfa4..c3974e1 100644 --- a/test/unit/app/layout/mail-shell.test.tsx +++ b/test/unit/app/layout/mail-shell.test.tsx @@ -102,12 +102,12 @@ describe("mail shell", () => { expect(html.indexOf("Settings")).toBeLessThan(html.indexOf("Open profile menu")); }); - it("combines the compact folder label and message count in one list header", () => { + it("combines the compact folder label and conversation count in one list header", () => { const html = renderToStaticMarkup( undefined} onRefresh={() => undefined} @@ -116,7 +116,7 @@ describe("mail shell", () => { ); expect(html).toContain('Inbox'); - expect(html).toContain(''); + expect(html).toContain(''); expect(html).not.toContain("Navigation"); }); diff --git a/test/unit/app/messages/conversation-reader.test.tsx b/test/unit/app/messages/conversation-reader.test.tsx index fcabad4..78d5602 100644 --- a/test/unit/app/messages/conversation-reader.test.tsx +++ b/test/unit/app/messages/conversation-reader.test.tsx @@ -5,7 +5,10 @@ import { InboxPage } from "@/features/inbox/inbox-page"; import { ConversationMessages } from "@/features/messages/conversation-messages"; import { MessageDetail } from "@/features/messages/message-detail"; import { MessageListItem } from "@/features/messages/message-list-item"; -import type { MessageDetail as MessageDetailType, MessageSummary } from "@/features/messages/types"; +import type { + ConversationSummary, + MessageDetail as MessageDetailType +} from "@/features/messages/types"; const firstMessage: MessageDetailType = { id: "msg_1", @@ -52,6 +55,14 @@ const secondMessage: MessageDetailType = { createdAt: "2026-07-27T14:05:00.000Z" }; +const conversation: ConversationSummary = { + ...secondMessage, + hasAttachments: false, + isStarred: false, + messageCount: 2, + unreadCount: 1 +}; + describe("conversation reader", () => { it("renders the complete thread and keeps Reply and Forward at the bottom", () => { const html = renderToStaticMarkup( @@ -70,16 +81,15 @@ describe("conversation reader", () => { expect(html).toContain(">Forward<"); expect(html).toContain('aria-label="Back to messages"'); expect(html).not.toContain('aria-label="Reply"'); - expect(html).toContain('aria-label="Archive"'); + expect(html).toContain('aria-label="Archive conversation"'); }); it("uses list-only and conversation-only compact states", () => { - const summary: MessageSummary = firstMessage; const listHtml = renderToStaticMarkup( undefined} onRefresh={() => undefined} @@ -89,9 +99,9 @@ describe("conversation reader", () => { const conversationHtml = renderToStaticMarkup( undefined} onRefresh={() => undefined} onSelect={() => undefined} @@ -108,22 +118,26 @@ describe("conversation reader", () => { it("labels the unread indicator and removes it once the message is read", () => { const unreadHtml = renderToStaticMarkup( undefined} /> ); const readHtml = renderToStaticMarkup( undefined} /> ); expect(unreadHtml).toContain('aria-label="Unread"'); + expect(unreadHtml).toContain('title="2 messages"'); + expect(unreadHtml).toContain("2 messages"); expect(readHtml).not.toContain('aria-label="Unread"'); }); diff --git a/test/unit/worker/features/messages/threading.test.ts b/test/unit/worker/features/messages/threading.test.ts new file mode 100644 index 0000000..de25a3c --- /dev/null +++ b/test/unit/worker/features/messages/threading.test.ts @@ -0,0 +1,53 @@ +import { describe, expect, it, vi } from "vitest"; + +vi.mock("@worker/db/client", () => ({ + newId: vi.fn(() => "thr_new"), + nowIso: vi.fn(() => "2026-07-28T16:00:00.000Z") +})); + +import { resolveInboundThread } from "@worker/features/messages/threading"; + +describe("message threading", () => { + it("reuses the thread referenced by In-Reply-To before subject matching", async () => { + const first = vi.fn(async () => ({ thread_id: "thr_existing" })); + const run = vi.fn(async () => ({ success: true })); + const prepare = vi.fn((sql: string) => ({ + bind: vi.fn(() => (sql.includes("WITH candidates") ? { first } : { run })) + })); + const db = { prepare } as unknown as D1Database; + + const threadId = await resolveInboundThread(db, { + inReplyTo: "", + lastMessageAt: "2026-07-28T15:00:00.000Z", + mailboxId: "mbx_1", + references: [""], + subject: "Re: Reused subject" + }); + + expect(threadId).toBe("thr_existing"); + expect(prepare.mock.calls[0]?.[0]).toContain("messages.message_id = candidates.value"); + expect(first).toHaveBeenCalledOnce(); + expect(run).toHaveBeenCalledOnce(); + }); + + it("creates a new thread when headers do not reference a stored message", async () => { + const first = vi.fn(async () => null); + const run = vi.fn(async () => ({ success: true })); + const prepare = vi.fn((sql: string) => ({ + bind: vi.fn(() => (sql.includes("WITH candidates") ? { first } : { run })) + })); + const db = { prepare } as unknown as D1Database; + + const threadId = await resolveInboundThread(db, { + inReplyTo: null, + lastMessageAt: "2026-07-28T15:00:00.000Z", + mailboxId: "mbx_1", + references: [], + subject: "Repeated subject" + }); + + expect(threadId).toBe("thr_new"); + expect(prepare).toHaveBeenCalledOnce(); + expect(prepare.mock.calls[0]?.[0]).toContain("INSERT INTO threads"); + }); +}); diff --git a/test/unit/worker/features/send/send-service.test.ts b/test/unit/worker/features/send/send-service.test.ts index 9ba01d9..4da7d81 100644 --- a/test/unit/worker/features/send/send-service.test.ts +++ b/test/unit/worker/features/send/send-service.test.ts @@ -13,21 +13,24 @@ vi.mock("@worker/features/mailboxes/address-queries", () => ({ })); vi.mock("@worker/features/messages/queries", () => ({ - ensureThread: vi.fn(), getMessageDetail: vi.fn(), getMessageHtmlKey: vi.fn(), insertAttachment: vi.fn(), insertMessage: vi.fn() })); +vi.mock("@worker/features/messages/threading", () => ({ + createThread: vi.fn(), + touchThread: vi.fn() +})); import { findMailboxForSending } from "@worker/features/mailboxes/queries"; import { - ensureThread, getMessageDetail, getMessageHtmlKey, insertAttachment, insertMessage } from "@worker/features/messages/queries"; +import { createThread, touchThread } from "@worker/features/messages/threading"; import { replyToMessage, sendNewMessage } from "@worker/features/send/service"; import type { WorkerEnv } from "@worker/lib/env"; @@ -81,7 +84,8 @@ describe("send service", () => { beforeEach(() => { vi.resetAllMocks(); vi.mocked(findMailboxForSending).mockResolvedValue(mailbox); - vi.mocked(ensureThread).mockResolvedValue("thread-1"); + vi.mocked(createThread).mockResolvedValue("thread-1"); + vi.mocked(touchThread).mockResolvedValue(); vi.mocked(insertMessage).mockResolvedValue(sentSummary); }); @@ -108,6 +112,7 @@ describe("send service", () => { env.DB, expect.objectContaining({ messageId: "" }) ); + expect(createThread).toHaveBeenCalledWith(env.DB, "Hello", "2026-07-10T00:00:00.000Z"); }); it("keeps only allowlisted threading headers on replies", async () => { @@ -166,6 +171,8 @@ describe("send service", () => { to: ["alternate@example.com"] }) ); + expect(createThread).not.toHaveBeenCalled(); + expect(touchThread).toHaveBeenCalledWith(env.DB, "thread-1", "2026-07-10T00:00:00.000Z"); expect(put).toHaveBeenCalledWith("sent/2026-07-10/html-1.html", quotedHtml, { httpMetadata: { contentType: "text/html; charset=utf-8" } }); diff --git a/worker/email/store-email.ts b/worker/email/store-email.ts index 01c4606..637eb75 100644 --- a/worker/email/store-email.ts +++ b/worker/email/store-email.ts @@ -1,12 +1,8 @@ import { newId, nowIso } from "../db/client"; import { findAddressIdentity } from "../features/mailboxes/address-queries"; import { findMailboxByAddress } from "../features/mailboxes/queries"; -import { - ensureThread, - getMessageDetail, - insertAttachment, - insertMessage -} from "../features/messages/queries"; +import { getMessageDetail, insertAttachment, insertMessage } from "../features/messages/queries"; +import { resolveInboundThread } from "../features/messages/threading"; import type { MessageDetail, MessageSummary } from "../features/messages/types"; import { attachmentBody, attachmentSize } from "./attachments"; @@ -57,7 +53,13 @@ export async function storeInboundEmail( mailboxId: mailbox?.id ?? null, parsed: input.parsed }); - const threadId = await ensureThread(db, input.parsed.subject, timestamp); + const threadId = await resolveInboundThread(db, { + inReplyTo: input.parsed.inReplyTo, + lastMessageAt: timestamp, + mailboxId: plan.mailboxId, + references: input.parsed.references, + subject: input.parsed.subject + }); const message = await insertMessage(db, { threadId, mailboxId: plan.mailboxId, diff --git a/worker/features/messages/conversation-queries.ts b/worker/features/messages/conversation-queries.ts new file mode 100644 index 0000000..9199937 --- /dev/null +++ b/worker/features/messages/conversation-queries.ts @@ -0,0 +1,194 @@ +import { nowIso } from "../../db/client"; +import { AppError } from "../../lib/errors"; + +import type { MessageAction } from "./actions"; +import type { ConversationFolder, ConversationRow, ConversationSummary } from "./types"; + +export type ListConversationFilters = { + folder?: ConversationFolder | undefined; + limit?: number | undefined; + mailboxId?: string | undefined; + mailboxIds: string[]; + search?: string | undefined; +}; + +export async function listConversations( + db: D1Database, + filters: ListConversationFilters +): Promise { + if (filters.mailboxIds.length === 0) { + return []; + } + + const accessibleWhere = `messages.mailbox_id IN (${filters.mailboxIds.map(() => "?").join(", ")})`; + const params: Array = [...filters.mailboxIds]; + + const eligibilityWhere: string[] = []; + if (filters.mailboxId) { + eligibilityWhere.push("accessible.mailbox_id = ?"); + params.push(filters.mailboxId); + } + if (filters.folder === "starred") { + eligibilityWhere.push("accessible.starred_at IS NOT NULL"); + } else if (filters.folder) { + eligibilityWhere.push("accessible.folder = ?"); + params.push(filters.folder); + } + if (filters.search) { + eligibilityWhere.push( + `(accessible.subject LIKE ? OR accessible.from_address LIKE ? + OR accessible.to_json LIKE ? OR accessible.snippet LIKE ? OR accessible.text_body LIKE ?)` + ); + const like = `%${filters.search}%`; + params.push(like, like, like, like, like); + } + + const limit = Math.min(Math.max(filters.limit ?? 100, 1), 100); + params.push(limit); + const result = await db + .prepare( + `WITH accessible AS ( + SELECT messages.*, + COALESCE(messages.received_at, messages.sent_at, messages.created_at) AS activity_at + FROM messages + WHERE ${accessibleWhere} + ), + eligible_threads AS ( + SELECT DISTINCT accessible.thread_id + FROM accessible + ${eligibilityWhere.length ? `WHERE ${eligibilityWhere.join(" AND ")}` : ""} + ), + ranked AS ( + SELECT accessible.*, + ROW_NUMBER() OVER ( + PARTITION BY accessible.thread_id + ORDER BY accessible.activity_at DESC, accessible.id DESC + ) AS thread_position + FROM accessible + JOIN eligible_threads ON eligible_threads.thread_id = accessible.thread_id + ), + aggregates AS ( + SELECT accessible.thread_id, + COUNT(*) AS message_count, + SUM( + CASE WHEN accessible.direction = 'inbound' AND accessible.read_at IS NULL + THEN 1 ELSE 0 END + ) AS unread_count, + MAX(CASE WHEN accessible.starred_at IS NOT NULL THEN 1 ELSE 0 END) AS is_starred, + MAX(accessible.has_attachments) AS has_thread_attachments + FROM accessible + JOIN eligible_threads ON eligible_threads.thread_id = accessible.thread_id + GROUP BY accessible.thread_id + ) + SELECT ranked.*, aggregates.message_count, aggregates.unread_count, + aggregates.is_starred, aggregates.has_thread_attachments + FROM ranked + JOIN aggregates ON aggregates.thread_id = ranked.thread_id + WHERE ranked.thread_position = 1 + ORDER BY ranked.activity_at DESC, ranked.id DESC + LIMIT ?` + ) + .bind(...params) + .all(); + + return result.results.map(mapConversationSummary); +} + +export async function updateConversationAction( + db: D1Database, + input: { + action: MessageAction; + activeFolder: ConversationFolder; + mailboxIds: string[]; + messageId: string; + } +): Promise<{ affected: number; threadId: string }> { + const selected = await db + .prepare("SELECT thread_id FROM messages WHERE id = ?") + .bind(input.messageId) + .first<{ thread_id: string }>(); + if (!selected) { + throw new AppError("MESSAGE_NOT_FOUND", "Message not found.", 404); + } + if (input.mailboxIds.length === 0) { + throw new AppError("MAILBOX_FORBIDDEN", "You do not have access to this mailbox.", 403); + } + + const timestamp = nowIso(); + const mailboxPlaceholders = input.mailboxIds.map(() => "?").join(", "); + const where = [`thread_id = ?`, `mailbox_id IN (${mailboxPlaceholders})`]; + const bindings: Array = [selected.thread_id, ...input.mailboxIds]; + let set: string; + + switch (input.action) { + case "read": + set = "read_at = ?, updated_at = ?"; + bindings.unshift(timestamp, timestamp); + where.push("direction = 'inbound'"); + break; + case "unread": + set = "read_at = NULL, updated_at = ?"; + bindings.unshift(timestamp); + where.push("direction = 'inbound'"); + break; + case "star": + set = "starred_at = ?, updated_at = ?"; + bindings.unshift(timestamp, timestamp); + break; + case "unstar": + set = "starred_at = NULL, updated_at = ?"; + bindings.unshift(timestamp); + break; + case "archive": + set = "folder = 'archived', archived_at = ?, updated_at = ?"; + bindings.unshift(timestamp, timestamp); + where.push("folder IN ('inbox', 'catchall')"); + break; + case "trash": + set = "folder = 'trash', trashed_at = ?, updated_at = ?"; + bindings.unshift(timestamp, timestamp); + if (input.activeFolder === "starred") { + where.push("starred_at IS NOT NULL"); + } else { + where.push("folder = ?"); + bindings.push(input.activeFolder); + } + break; + } + + const result = await db + .prepare(`UPDATE messages SET ${set} WHERE ${where.join(" AND ")}`) + .bind(...bindings) + .run(); + return { affected: result.meta.changes, threadId: selected.thread_id }; +} + +function mapConversationSummary(row: ConversationRow): ConversationSummary { + return { + id: row.id, + threadId: row.thread_id, + mailboxId: row.mailbox_id, + direction: row.direction, + folder: row.folder, + fromAddress: row.from_address, + to: parseJsonList(row.to_json), + subject: row.subject, + snippet: row.snippet, + receivedAt: row.received_at, + sentAt: row.sent_at, + readAt: row.read_at, + starredAt: row.starred_at, + hasAttachments: row.has_thread_attachments === 1, + createdAt: row.created_at, + isStarred: row.is_starred === 1, + messageCount: row.message_count, + unreadCount: row.unread_count + }; +} + +function parseJsonList(value: string): string[] { + const parsed = JSON.parse(value) as unknown; + return Array.isArray(parsed) + ? parsed.filter((item): item is string => typeof item === "string") + : []; +} diff --git a/worker/features/messages/conversation-routes.ts b/worker/features/messages/conversation-routes.ts new file mode 100644 index 0000000..7b8bb1f --- /dev/null +++ b/worker/features/messages/conversation-routes.ts @@ -0,0 +1,61 @@ +import { Hono } from "hono"; +import { z } from "zod"; + +import { accessibleMailboxIds, requireMailboxAccess } from "../../auth/mailbox-access"; +import { requireAuthContext } from "../../auth/session"; +import type { HonoApp } from "../../lib/env"; +import { parseWith } from "../../lib/validation"; + +import type { MessageAction } from "./actions"; +import { listConversations, updateConversationAction } from "./conversation-queries"; +import { getMessageMailboxId } from "./queries"; +import { conversationFolders } from "./types"; + +export const conversationRoutes = new Hono(); + +const actions: readonly MessageAction[] = ["read", "unread", "star", "unstar", "archive", "trash"]; +const folderSchema = z.enum(conversationFolders); +const actionBodySchema = z.object({ folder: folderSchema }); + +conversationRoutes.get("/", async (c) => { + const auth = await requireAuthContext(c.env, c.req.raw); + const mailboxIds = await accessibleMailboxIds(c.env.DB, auth.user.id, auth.user.role, "read"); + const folder = parseWith(folderSchema.optional(), c.req.query("folder")); + return c.json( + await listConversations(c.env.DB, { + folder, + mailboxId: c.req.query("mailboxId"), + search: c.req.query("search"), + mailboxIds + }) + ); +}); + +for (const action of actions) { + conversationRoutes.post(`/:id/${action}`, async (c) => { + const auth = await requireAuthContext(c.env, c.req.raw); + const requiredAccess = action === "read" || action === "unread" ? "read" : "agent"; + await requireMailboxAccess( + c.env.DB, + auth.user.id, + auth.user.role, + await getMessageMailboxId(c.env.DB, c.req.param("id")), + requiredAccess + ); + const mailboxIds = await accessibleMailboxIds( + c.env.DB, + auth.user.id, + auth.user.role, + requiredAccess + ); + const body = parseWith(actionBodySchema, await c.req.json().catch(() => ({}))); + return c.json( + await updateConversationAction(c.env.DB, { + action, + activeFolder: body.folder, + mailboxIds, + messageId: c.req.param("id") + }) + ); + }); +} diff --git a/worker/features/messages/queries.ts b/worker/features/messages/queries.ts index 9197d77..f1ec237 100644 --- a/worker/features/messages/queries.ts +++ b/worker/features/messages/queries.ts @@ -2,7 +2,6 @@ import { newId, nowIso } from "../../db/client"; import { AppError } from "../../lib/errors"; import type { MessageAction } from "./actions"; import { buildMessageActionPatch } from "./actions"; -import { normalizeSubject } from "./headers"; import type { AttachmentRow, InsertAttachmentInput, @@ -26,37 +25,6 @@ export type ListMessageFilters = { mailboxIds?: string[] | undefined; }; -export async function ensureThread( - db: D1Database, - subject: string, - timestamp: string -): Promise { - const subjectNormalized = normalizeSubject(subject); - const existing = await db - .prepare("SELECT id FROM threads WHERE subject_normalized = ? LIMIT 1") - .bind(subjectNormalized) - .first<{ id: string }>(); - - if (existing) { - await db - .prepare("UPDATE threads SET last_message_at = ?, updated_at = ? WHERE id = ?") - .bind(timestamp, timestamp, existing.id) - .run(); - return existing.id; - } - - const id = newId("thr"); - await db - .prepare( - `INSERT INTO threads (id, subject_normalized, last_message_at, created_at, updated_at) - VALUES (?, ?, ?, ?, ?)` - ) - .bind(id, subjectNormalized, timestamp, timestamp, timestamp) - .run(); - - return id; -} - export async function insertMessage( db: D1Database, input: InsertMessageInput diff --git a/worker/features/messages/threading.ts b/worker/features/messages/threading.ts new file mode 100644 index 0000000..90db973 --- /dev/null +++ b/worker/features/messages/threading.ts @@ -0,0 +1,95 @@ +import { newId, nowIso } from "../../db/client"; + +import { normalizeSubject, parseReferences } from "./headers"; + +const maxReferenceCandidates = 32; + +export async function createThread( + db: D1Database, + subject: string, + lastMessageAt: string +): Promise { + const id = newId("thr"); + const timestamp = nowIso(); + await db + .prepare( + `INSERT INTO threads (id, subject_normalized, last_message_at, created_at, updated_at) + VALUES (?, ?, ?, ?, ?)` + ) + .bind(id, normalizeSubject(subject), lastMessageAt, timestamp, timestamp) + .run(); + return id; +} + +export async function resolveInboundThread( + db: D1Database, + input: { + inReplyTo: string | null; + lastMessageAt: string; + mailboxId: string | null; + references: string[]; + subject: string; + } +): Promise { + const existing = await findReferencedThread(db, input); + if (!existing) { + return createThread(db, input.subject, input.lastMessageAt); + } + + await touchThread(db, existing, input.lastMessageAt); + return existing; +} + +export async function touchThread( + db: D1Database, + threadId: string, + lastMessageAt: string +): Promise { + await db + .prepare( + `UPDATE threads + SET last_message_at = MAX(last_message_at, ?), updated_at = ? + WHERE id = ?` + ) + .bind(lastMessageAt, nowIso(), threadId) + .run(); +} + +async function findReferencedThread( + db: D1Database, + input: { + inReplyTo: string | null; + mailboxId: string | null; + references: string[]; + } +): Promise { + const candidates = unique([ + ...parseReferences(input.inReplyTo), + ...[...input.references].reverse() + ]).slice(0, maxReferenceCandidates); + if (candidates.length === 0) { + return null; + } + + const values = candidates.map(() => "(?, ?)").join(", "); + const bindings = candidates.flatMap((value, priority) => [value, priority]); + const row = await db + .prepare( + `WITH candidates(value, priority) AS (VALUES ${values}) + SELECT messages.thread_id + FROM candidates + JOIN messages + ON messages.message_id = candidates.value OR messages.id = candidates.value + ORDER BY candidates.priority, + CASE WHEN messages.mailbox_id IS ? THEN 0 ELSE 1 END, + COALESCE(messages.received_at, messages.sent_at, messages.created_at) DESC + LIMIT 1` + ) + .bind(...bindings, input.mailboxId) + .first<{ thread_id: string }>(); + return row?.thread_id ?? null; +} + +function unique(values: string[]): string[] { + return [...new Set(values.map((value) => value.trim()).filter(Boolean))]; +} diff --git a/worker/features/messages/types.ts b/worker/features/messages/types.ts index efe7883..586aa7c 100644 --- a/worker/features/messages/types.ts +++ b/worker/features/messages/types.ts @@ -1,8 +1,17 @@ export const messageFolders = ["inbox", "sent", "drafts", "archived", "trash", "catchall"] as const; export const messageDirections = ["inbound", "outbound"] as const; +export const conversationFolders = [ + "inbox", + "sent", + "starred", + "archived", + "trash", + "catchall" +] as const; export type MessageFolder = (typeof messageFolders)[number]; export type MessageDirection = (typeof messageDirections)[number]; +export type ConversationFolder = (typeof conversationFolders)[number]; export type StoredAttachment = { id: string; @@ -45,6 +54,12 @@ export type MessageDetail = MessageSummary & { attachments: StoredAttachment[]; }; +export type ConversationSummary = MessageSummary & { + isStarred: boolean; + messageCount: number; + unreadCount: number; +}; + export type MessageRow = { id: string; thread_id: string; @@ -76,6 +91,13 @@ export type MessageRow = { delivered_to_address: string | null; }; +export type ConversationRow = Omit & { + has_thread_attachments: number; + is_starred: number; + message_count: number; + unread_count: number; +}; + export type AttachmentRow = { id: string; message_id: string; diff --git a/worker/features/send/service.ts b/worker/features/send/service.ts index 34291a1..75f2d3f 100644 --- a/worker/features/send/service.ts +++ b/worker/features/send/service.ts @@ -8,12 +8,12 @@ import { ensureReplySubject } from "../messages/headers"; import { sanitizeQuotedMessageHtml } from "../messages/html-sanitizer"; import { isSafeInlineImage } from "../messages/inline-media"; import { - ensureThread, getMessageDetail, getMessageHtmlKey, insertAttachment, insertMessage } from "../messages/queries"; +import { createThread, touchThread } from "../messages/threading"; import type { MessageSummary, StoredAttachment } from "../messages/types"; import { buildReplyBody } from "./reply-body"; @@ -41,6 +41,7 @@ export async function sendNewMessage( ...(input.html ? { html: input.html } : {}), ...(attachments.length ? { attachments: attachments.map(asEmailAttachment) } : {}) }); + const threadId = await createThread(env.DB, input.subject, timestamp); return storeSentMessage(env, { ...input, @@ -49,6 +50,7 @@ export async function sendNewMessage( references: [], sentAt: timestamp, subject: input.subject, + threadId, storedAttachments: attachments, draftId: input.draftId ?? null, userId: userId ?? null @@ -113,6 +115,7 @@ export async function replyToMessage( messageId: sendResult.messageId, references, sentAt: timestamp, + threadId: original.threadId, storedAttachments: outgoingAttachments, draftId: input.draftId ?? null, userId: userId ?? null @@ -143,6 +146,7 @@ async function storeSentMessage( messageId: string; references: string[]; sentAt: string; + threadId: string; storedAttachments: StoredOutgoingAttachment[]; draftId: string | null; userId: string | null; @@ -160,10 +164,10 @@ async function storeSentMessage( }); } - const threadId = await ensureThread(env.DB, input.subject, input.sentAt); + await touchThread(env.DB, input.threadId, input.sentAt); const sendingIdentity = await findAddressIdentity(env.DB, input.from, "send"); const message = await insertMessage(env.DB, { - threadId, + threadId: input.threadId, mailboxId: mailbox.id, direction: "outbound", folder: "sent", diff --git a/worker/routes/index.ts b/worker/routes/index.ts index cbbcf89..0fc1ab6 100644 --- a/worker/routes/index.ts +++ b/worker/routes/index.ts @@ -6,6 +6,7 @@ import { domainRoutes } from "../features/domains/routes"; import { draftRoutes } from "../features/drafts/routes"; import { mailboxAccessRoutes } from "../features/mailbox-access/routes"; import { mailboxRoutes } from "../features/mailboxes/routes"; +import { conversationRoutes } from "../features/messages/conversation-routes"; import { attachmentRoutes, messageRoutes } from "../features/messages/routes"; import { operationRoutes } from "../features/operations/routes"; import { sendRoutes } from "../features/send/routes"; @@ -53,6 +54,7 @@ apiRoutes.route("/api/mailbox-grants", mailboxAccessRoutes); apiRoutes.route("/api/sessions", sessionControlRoutes); apiRoutes.route("/api/operations", operationRoutes); apiRoutes.route("/api/mailboxes", mailboxRoutes); +apiRoutes.route("/api/conversations", conversationRoutes); apiRoutes.route("/api/messages", messageRoutes); apiRoutes.route("/api/attachments", attachmentRoutes); apiRoutes.route("/api/users", userRoutes);