From ab9bce6203e751c4077735217e126d9737d062b1 Mon Sep 17 00:00:00 2001 From: hrsvrn Date: Wed, 1 Jul 2026 16:18:52 +0530 Subject: [PATCH] feat(email): drafts, search, bulk read/unread, configurable backfill Gmail client enhancements in apps/x, bundling four independent features: - Drafts: save/autosave, update, delete, and list Gmail drafts. The composer autosaves to a real Gmail draft (debounced ~1.5s) while typing, reuses the thread's existing draft so edits update in place, flushes a final save on close, and deletes the draft on discard. Adds isDraft/draftId to the shared thread types. New core helpers: saveThreadDraft, deleteThreadDraft, listDraftThreads, buildDraftSnapshot, buildRawMimeMessage. - Search: searchThreads(query, {limit}) backed by an on-disk snapshot cache (read/writeSearchSnapshot). Extracts parseThreadSnapshot as the shared parse core reused by both the cache-building sync and search. - Read state: markThreadRead now takes a `read` flag so it toggles read/unread; new markSectionRead marks a whole section (important/other) read/unread and returns the affected count. - Backfill: the onboarding/recovery sync is now bounded by a configurable thread COUNT instead of a fixed 7-day window. New gmail_sync_config.ts (getMaxEmails/setMaxEmails) backed by ~/.rowboat/config/gmail_sync.json, default 500, clamped to 1-5000; seeded on first run. New IPC channels: gmail:saveDraft, gmail:deleteDraft, gmail:getDrafts, gmail:search, gmail:markSectionRead (plus a `read` field on gmail:markThreadRead). --- apps/x/apps/main/src/ipc.ts | 19 +- .../renderer/src/components/email-view.tsx | 488 ++++++++++++- apps/x/packages/core/src/config/config.ts | 11 + .../core/src/config/gmail_sync_config.ts | 61 ++ .../packages/core/src/knowledge/sync_gmail.ts | 674 ++++++++++++++---- apps/x/packages/shared/src/blocks.ts | 5 + apps/x/packages/shared/src/ipc.ts | 56 +- 7 files changed, 1149 insertions(+), 165 deletions(-) create mode 100644 apps/x/packages/core/src/config/gmail_sync_config.ts diff --git a/apps/x/apps/main/src/ipc.ts b/apps/x/apps/main/src/ipc.ts index 3ed9f6a0..ae732230 100644 --- a/apps/x/apps/main/src/ipc.ts +++ b/apps/x/apps/main/src/ipc.ts @@ -69,7 +69,7 @@ import { summarizeMeeting } from '@x/core/dist/knowledge/summarize_meeting.js'; import { getAccessToken } from '@x/core/dist/auth/tokens.js'; import { getRowboatConfig } from '@x/core/dist/config/rowboat.js'; import { runLiveNoteAgent } from '@x/core/dist/knowledge/live-note/runner.js'; -import { listImportantThreads, listEverythingElseThreads, saveMessageBodyHeight, triggerSync as triggerGmailSync, sendThreadReply, archiveThread, trashThread, markThreadRead, getAccountEmail, getAccountName, getConnectionStatus as getGmailConnectionStatus } from '@x/core/dist/knowledge/sync_gmail.js'; +import { listImportantThreads, listEverythingElseThreads, saveMessageBodyHeight, triggerSync as triggerGmailSync, sendThreadReply, saveThreadDraft, deleteThreadDraft, listDraftThreads, searchThreads, archiveThread, trashThread, markThreadRead, markSectionRead, getAccountEmail, getAccountName, getConnectionStatus as getGmailConnectionStatus } from '@x/core/dist/knowledge/sync_gmail.js'; import { searchContacts as searchGmailContacts, warmContactIndex } from '@x/core/dist/knowledge/gmail_contacts.js'; import { searchSentContacts, warmSentContacts } from '@x/core/dist/knowledge/gmail_sent_contacts.js'; import { getGoogleDocsConnectionStatus, importGoogleDoc, syncGoogleDocDown, syncGoogleDocUp, getGoogleDocLink } from '@x/core/dist/knowledge/google_docs.js'; @@ -740,6 +740,18 @@ export function setupIpcHandlers() { 'gmail:sendReply': async (_event, args) => { return sendThreadReply(args); }, + 'gmail:saveDraft': async (_event, args) => { + return saveThreadDraft(args); + }, + 'gmail:deleteDraft': async (_event, args) => { + return deleteThreadDraft(args.draftId); + }, + 'gmail:getDrafts': async () => { + return listDraftThreads(); + }, + 'gmail:search': async (_event, args) => { + return searchThreads(args.query, { limit: args.limit }); + }, 'gmail:getConnectionStatus': async () => { return getGmailConnectionStatus(); }, @@ -756,7 +768,10 @@ export function setupIpcHandlers() { return trashThread(args.threadId); }, 'gmail:markThreadRead': async (_event, args) => { - return markThreadRead(args.threadId); + return markThreadRead(args.threadId, args.read); + }, + 'gmail:markSectionRead': async (_event, args) => { + return markSectionRead(args.section, args.read); }, 'gmail:saveMessageHeight': async (_event, args) => { saveMessageBodyHeight(args.threadId, args.messageId, args.height); diff --git a/apps/x/apps/renderer/src/components/email-view.tsx b/apps/x/apps/renderer/src/components/email-view.tsx index 8815723f..f469e9bd 100644 --- a/apps/x/apps/renderer/src/components/email-view.tsx +++ b/apps/x/apps/renderer/src/components/email-view.tsx @@ -11,6 +11,7 @@ import { useTheme } from '@/contexts/theme-context' import { SettingsDialog } from '@/components/settings-dialog' import { Button } from '@/components/ui/button' import { Input } from '@/components/ui/input' +import { Switch } from '@/components/ui/switch' import { Dialog, DialogContent, DialogTitle } from '@/components/ui/dialog' type GmailThread = blocks.GmailThread @@ -166,6 +167,11 @@ function buildRecipients( selfEmail: string, ): { to: string[]; cc: string[] } { if (mode === 'forward') return { to: [], cc: [] } + // Editing an existing draft: recipients are whatever the draft already has. + if (mode === 'draft') { + const draftMsg = latestMessage(thread) + return { to: splitAddresses(draftMsg?.to), cc: splitAddresses(draftMsg?.cc) } + } const latest = latestMessage(thread) const self = selfEmail.toLowerCase() @@ -195,6 +201,7 @@ function buildRecipients( // Subject line for a reply ("Re: …") or forward ("Fwd: …"), avoiding double prefixes. function composeSubject(mode: ComposeMode, rawSubject?: string): string { const raw = (rawSubject || '').trim() + if (mode === 'draft') return raw // keep the draft's own subject verbatim if (mode === 'forward') return /^fwd:/i.test(raw) ? raw : `Fwd: ${raw}`.trim() return /^re:/i.test(raw) ? raw : `Re: ${raw}`.trim() } @@ -526,7 +533,7 @@ function MessageAttachments({ attachments }: { attachments: NonNullable = { key: 'longer', label: 'Longer', instruction: 'Rewrite this email to be more detailed and thorough.' }, ] +// Debounce before autosaving the composer to a Gmail draft after the last edit. +const DRAFT_AUTOSAVE_MS = 1500 + +// Shape of the gmail:saveDraft request, kept here so we can stash a snapshot in +// a ref for the close/unmount flush without re-reading a torn-down editor. +type DraftPayload = { + draftId?: string + threadId?: string + to?: string + cc?: string + bcc?: string + subject: string + bodyHtml: string + bodyText: string + inReplyTo?: string + references?: string + attachments?: Array<{ filename: string; mimeType: string; contentBase64: string }> +} + // Composer for replies, forwards, and (mode 'new') from-scratch emails. With a // thread it renders as an inline card under the thread; in 'new' mode it has no // thread and renders as a centered modal with the AI writing bar. @@ -992,7 +1018,9 @@ const ComposeBox = memo(function ComposeBox({ onClose: () => void }) { const isNew = mode === 'new' - const latest = thread ? latestMessage(thread) : undefined + // Drafts and new messages share the full-modal layout (subject line, AI bar). + const isModal = isNew || mode === 'draft' + const latest = useMemo(() => (thread ? latestMessage(thread) : undefined), [thread]) const initialRecipients = useMemo( () => (thread ? buildRecipients(mode, thread, selfEmail) : { to: [], cc: [] }), [mode, thread, selfEmail], @@ -1004,11 +1032,16 @@ const ComposeBox = memo(function ComposeBox({ const [showCc, setShowCc] = useState(initialRecipients.cc.length > 0) const [showBcc, setShowBcc] = useState(false) const [subject, setSubject] = useState(() => (thread ? composeSubject(mode, thread.subject) : '')) - const modeLabel = isNew ? 'New message' : mode === 'forward' ? 'Forward' : mode === 'replyAll' ? 'Reply All' : 'Reply' + const modeLabel = mode === 'draft' ? 'Draft' : isNew ? 'New message' : mode === 'forward' ? 'Forward' : mode === 'replyAll' ? 'Reply All' : 'Reply' const initialContent = useMemo(() => { if (!thread) return '' if (mode === 'forward') return buildForwardedContent(thread) + // For a saved draft, reopen the exact HTML we stored so formatting survives. + if (mode === 'draft') { + const draftMsg = latestMessage(thread) + if (draftMsg?.bodyHtml) return draftMsg.bodyHtml + } // Gmail-side draft (user's own work) wins over the AI-generated draft. const source = stripQuotedReplyText(thread.gmail_draft || thread.draft_response || '') if (!source) return '' @@ -1023,7 +1056,7 @@ const ComposeBox = memo(function ComposeBox({ StarterKit.configure({ link: false }), Link.configure({ openOnClick: false, autolink: true }), Placeholder.configure({ - placeholder: isNew || mode === 'forward' ? 'Write a message…' : 'Write your reply…', + placeholder: isModal || mode === 'forward' ? 'Write a message…' : 'Write your reply…', }), ], editorProps: { @@ -1255,6 +1288,155 @@ const ComposeBox = memo(function ComposeBox({ setAttachments((prev) => prev.filter((a) => a.id !== id)) } + // ── Draft autosave ───────────────────────────────────────────────────────── + // Keep a real Gmail draft in sync with the composer while the user types + // (debounced) and flush a final save on close, so closing keeps the work in + // Gmail's Drafts folder (synced to every device) instead of discarding it. + // Empty/whitespace drafts are skipped. The core saveThreadDraft reuses an + // existing thread draft and tracks the id so edits update in place. + // Seeded when editing an existing draft so the first save updates it in place. + const draftIdRef = useRef(thread?.draftId) + const lastPayloadRef = useRef(null) + const savingRef = useRef(false) // a saveDraft IPC is in flight + const pendingRef = useRef(false) // edits arrived mid-flight; save once more + const sentRef = useRef(false) // suppress autosave after a successful send + const discardedRef = useRef(false) // suppress autosave after discard + const closedRef = useRef(false) // close already handled; skip unmount flush + const dirtyRef = useRef(false) // user has edited since open + const fieldsMounted = useRef(false) // skip the field effect's initial run + const autosaveTimer = useRef(null) + const saveDraftNowRef = useRef<() => Promise>(undefined) + + const buildDraftPayload = useCallback((): DraftPayload | null => { + if (!editor || editor.isDestroyed) return null + const text = editor.getText().trim() + if (!text) return null // skip empty/whitespace drafts + const html = editor.getHTML() + const messageIds = (thread?.messages ?? []) + .map((m) => m.messageIdHeader) + .filter((v): v is string => Boolean(v)) + const references = messageIds.join(' ') + const inReplyTo = latest?.messageIdHeader + // Only replies stay on the thread; forwards and new emails start fresh. + const isThreaded = Boolean(thread) && mode !== 'forward' && !isNew + return { + draftId: draftIdRef.current, + threadId: isThreaded ? thread?.threadId : undefined, + to: toList.join(', '), + cc: ccList.length ? ccList.join(', ') : undefined, + bcc: bccList.length ? bccList.join(', ') : undefined, + subject: subject.trim() || (thread ? composeSubject(mode, thread.subject) : ''), + bodyHtml: html, + bodyText: text, + inReplyTo: isThreaded ? inReplyTo : undefined, + references: isThreaded ? references || undefined : undefined, + attachments: attachments.length + ? attachments.map(({ filename, mimeType, contentBase64 }) => ({ filename, mimeType, contentBase64 })) + : undefined, + } + }, [editor, thread, mode, isNew, toList, ccList, bccList, subject, attachments, latest]) + + const saveDraftNow = useCallback(async () => { + if (sentRef.current || discardedRef.current) return + if (savingRef.current) { pendingRef.current = true; return } + const payload = lastPayloadRef.current + if (!payload) return + savingRef.current = true + pendingRef.current = false + try { + payload.draftId = draftIdRef.current + const res = await window.ipc.invoke('gmail:saveDraft', payload) + if (res?.draftId && !discardedRef.current) draftIdRef.current = res.draftId + } catch { + // Autosave is best-effort; a failure just leaves the prior draft in place. + } finally { + savingRef.current = false + // Coalesce edits that landed mid-flight into one more save. + if (pendingRef.current && !sentRef.current && !discardedRef.current) { + pendingRef.current = false + void saveDraftNowRef.current?.() + } + } + }, []) + useEffect(() => { saveDraftNowRef.current = saveDraftNow }, [saveDraftNow]) + + const scheduleAutosave = useCallback(() => { + if (autosaveTimer.current) window.clearTimeout(autosaveTimer.current) + autosaveTimer.current = window.setTimeout(() => { void saveDraftNow() }, DRAFT_AUTOSAVE_MS) + }, [saveDraftNow]) + + // Wait (briefly, capped) for any in-flight autosave to settle so send/discard + // don't race it into a duplicate or orphaned draft. + const waitForSaveIdle = useCallback(async () => { + for (let guard = 0; savingRef.current && guard < 40; guard++) { + await new Promise((r) => window.setTimeout(r, 50)) + } + }, []) + + // Autosave on body edits (typing, AI insert, undo/redo). + useEffect(() => { + if (!editor) return + const onUpdate = () => { + dirtyRef.current = true + lastPayloadRef.current = buildDraftPayload() + scheduleAutosave() + } + editor.on('update', onUpdate) + return () => { editor.off('update', onUpdate) } + }, [editor, buildDraftPayload, scheduleAutosave]) + + // Autosave on header edits (recipients, subject, attachments), skipping mount. + useEffect(() => { + if (!fieldsMounted.current) { fieldsMounted.current = true; return } + dirtyRef.current = true + lastPayloadRef.current = buildDraftPayload() + scheduleAutosave() + }, [toList, ccList, bccList, subject, attachments, buildDraftPayload, scheduleAutosave]) + + // Safety net: if the composer is torn down without an explicit close (e.g. + // navigating away), still flush unsaved edits. closedRef guards the common + // path where handleClose already saved. + useEffect(() => { + return () => { + if (autosaveTimer.current) window.clearTimeout(autosaveTimer.current) + if (closedRef.current || sentRef.current || discardedRef.current || savingRef.current) return + if (!dirtyRef.current) return + const payload = lastPayloadRef.current + if (!payload) return + payload.draftId = draftIdRef.current + void window.ipc.invoke('gmail:saveDraft', payload).catch(() => {}) + } + }, []) + + // Close (X / click-away): keep the draft. Flush a final save when there are + // unsaved edits. If a save is already in flight it will persist the latest + // content, so we skip here to avoid creating a duplicate. + const handleClose = useCallback(() => { + closedRef.current = true + if (autosaveTimer.current) window.clearTimeout(autosaveTimer.current) + if (!sentRef.current && !discardedRef.current && !savingRef.current && dirtyRef.current) { + const payload = buildDraftPayload() + if (payload) { + payload.draftId = draftIdRef.current + void window.ipc.invoke('gmail:saveDraft', payload).catch(() => {}) + } + } + onClose() + }, [buildDraftPayload, onClose]) + + // Discard: delete the autosaved draft (if any), then close. + const handleDiscard = useCallback(() => { + discardedRef.current = true + closedRef.current = true + if (autosaveTimer.current) window.clearTimeout(autosaveTimer.current) + void (async () => { + await waitForSaveIdle() + const id = draftIdRef.current + if (id) await window.ipc.invoke('gmail:deleteDraft', { draftId: id }).catch(() => {}) + })() + onClose() + }, [onClose, waitForSaveIdle]) + const [sending, setSending] = useState(false) const sendInGmail = async () => { if (!editor || sending) return @@ -1279,7 +1461,12 @@ const ComposeBox = memo(function ComposeBox({ // Only replies stay on the thread; forwards and new emails start fresh. const isThreaded = Boolean(thread) && mode !== 'forward' && !isNew + // Stop autosave from racing the send (it would leave an orphaned draft), and + // let any in-flight save settle so we know the draft id to clean up. + sentRef.current = true + if (autosaveTimer.current) window.clearTimeout(autosaveTimer.current) setSending(true) + await waitForSaveIdle() try { const result = await window.ipc.invoke('gmail:sendReply', { threadId: isThreaded ? thread?.threadId : undefined, @@ -1296,12 +1483,22 @@ const ComposeBox = memo(function ComposeBox({ : undefined, }) if (result.error) { + sentRef.current = false // allow autosave to resume on a failed send toast(`Send failed: ${result.error}`, 'error') return } + // Gmail only auto-cleans drafts on a threaded send; remove any draft we + // autosaved for a brand-new message so it doesn't linger after sending. + const leftover = draftIdRef.current + if (leftover) { + draftIdRef.current = undefined + void window.ipc.invoke('gmail:deleteDraft', { draftId: leftover }).catch(() => {}) + } toast('Sent.', 'success') + closedRef.current = true onClose() } catch (err) { + sentRef.current = false toast(`Send failed: ${err instanceof Error ? err.message : String(err)}`, 'error') } finally { setSending(false) @@ -1419,7 +1616,7 @@ const ComposeBox = memo(function ComposeBox({ >{preset.label} ))} - {(isNew || mode === 'forward') && ( + {(isModal || mode === 'forward') && (
Subject {editor && } -
) - if (isNew) { + if (isModal) { return ( - { if (!open) onClose() }}> + { if (!open) handleClose() }}> @@ -1569,7 +1766,7 @@ const ComposeBox = memo(function ComposeBox({ variant="ghost" size="icon-xs" className="text-muted-foreground" - onClick={onClose} + onClick={handleClose} aria-label="Close compose" > @@ -1731,6 +1928,8 @@ const initialSectionState: SectionState = { // panels and coming back doesn't reload from scratch. let persistedImportant: SectionState | null = null let persistedOther: SectionState | null = null +// Last-loaded drafts, kept across EmailView remounts so reopening is instant. +let persistedDrafts: GmailThread[] | null = null function clearLoadingFlag(state: SectionState | null): SectionState { if (!state) return initialSectionState @@ -1765,10 +1964,86 @@ export function EmailView({ initialThreadId, threadIdVersion }: EmailViewProps = const [composeOpen, setComposeOpen] = useState(false) // Stable so the open composer isn't re-rendered on every inbox sync tick. const closeCompose = useCallback(() => setComposeOpen(false), []) + // Inbox vs Drafts. Drafts are fetched live (they're not in the inbox cache). + const [view, setView] = useState<'inbox' | 'drafts'>('inbox') + const [drafts, setDrafts] = useState(() => persistedDrafts ?? []) + const [draftsLoading, setDraftsLoading] = useState(false) + const [draftsError, setDraftsError] = useState(null) + const [editingDraft, setEditingDraft] = useState(null) + // Server-side search across the whole Gmail mailbox (results indexed locally). + const [searchResults, setSearchResults] = useState([]) + const [searching, setSearching] = useState(false) + const [searchError, setSearchError] = useState(null) + const searchEpoch = useRef(0) // Gmail sync uses the native Google OAuth connection. const [emailConnection, setEmailConnection] = useState(null) const [settingsOpen, setSettingsOpen] = useState(false) + const loadDrafts = useCallback(async () => { + setDraftsLoading(true) + setDraftsError(null) + try { + const res = await window.ipc.invoke('gmail:getDrafts', {}) + if (res.error) setDraftsError(res.error) + setDrafts(res.threads ?? []) + } catch (err) { + setDraftsError(err instanceof Error ? err.message : String(err)) + } finally { + setDraftsLoading(false) + } + }, []) + + // Load drafts when the Drafts view is opened. + useEffect(() => { + if (view === 'drafts') void loadDrafts() + }, [view, loadDrafts]) + + // Debounced full-mailbox search. Each keystroke bumps an epoch so stale + // responses are ignored; an empty query clears results. + useEffect(() => { + const q = query.trim() + searchEpoch.current += 1 + const epoch = searchEpoch.current + if (!q) { + setSearchResults([]) + setSearchError(null) + setSearching(false) + return + } + setSearching(true) + const handle = window.setTimeout(async () => { + try { + const res = await window.ipc.invoke('gmail:search', { query: q, limit: 100 }) + if (searchEpoch.current !== epoch) return + setSearchError(res.error ?? null) + setSearchResults(res.threads ?? []) + } catch (err) { + if (searchEpoch.current === epoch) setSearchError(err instanceof Error ? err.message : String(err)) + } finally { + if (searchEpoch.current === epoch) setSearching(false) + } + }, 400) + return () => window.clearTimeout(handle) + }, [query]) + + const deleteDraftAction = useCallback(async (thread: GmailThread) => { + const id = thread.draftId + if (!id) return + setDrafts((prev) => prev.filter((d) => d.draftId !== id)) + try { + await window.ipc.invoke('gmail:deleteDraft', { draftId: id }) + } catch (err) { + toast(`Could not delete draft: ${err instanceof Error ? err.message : String(err)}`, 'error') + void loadDrafts() + } + }, [loadDrafts]) + + // Closing the draft composer may have edited/sent/deleted it — refresh. + const closeDraftEditor = useCallback(() => { + setEditingDraft(null) + void loadDrafts() + }, [loadDrafts]) + useEffect(() => { let cancelled = false const check = async () => { @@ -1800,7 +2075,7 @@ export function EmailView({ initialThreadId, threadIdVersion }: EmailViewProps = useEffect(() => { const handleKeyDown = (e: KeyboardEvent) => { if (e.key !== 'n' || e.ctrlKey || e.metaKey || e.altKey || e.shiftKey || e.isComposing) return - if (composeOpen || settingsOpen) return + if (composeOpen || settingsOpen || editingDraft) return const target = e.target as HTMLElement | null if ( target && @@ -1816,10 +2091,11 @@ export function EmailView({ initialThreadId, threadIdVersion }: EmailViewProps = } document.addEventListener('keydown', handleKeyDown) return () => document.removeEventListener('keydown', handleKeyDown) - }, [composeOpen, settingsOpen]) + }, [composeOpen, settingsOpen, editingDraft]) useEffect(() => { persistedImportant = important }, [important]) useEffect(() => { persistedOther = other }, [other]) + useEffect(() => { persistedDrafts = drafts }, [drafts]) const setSection = useCallback((section: InboxSection, updater: (prev: SectionState) => SectionState) => { if (section === 'important') setImportant(updater) @@ -1846,20 +2122,39 @@ export function EmailView({ initialThreadId, threadIdVersion }: EmailViewProps = setOpenedThreadIds((prev) => prev.filter((id) => id !== threadId)) }, []) - const markThreadReadAction = useCallback(async (threadId: string) => { + const markThreadReadAction = useCallback(async (threadId: string, read: boolean = true) => { updateThreadInState(threadId, (t) => ({ ...t, - unread: false, - messages: t.messages.map((m) => ({ ...m, unread: false })), + unread: !read, + messages: t.messages.map((m) => ({ ...m, unread: !read })), })) try { - const result = await window.ipc.invoke('gmail:markThreadRead', { threadId }) + const result = await window.ipc.invoke('gmail:markThreadRead', { threadId, read }) if (!result.ok && result.error) console.warn('[Gmail] mark-read failed:', result.error) } catch (err) { console.warn('[Gmail] mark-read failed:', err) } }, [updateThreadInState]) + // Toggle the read state of a whole category — server-side across the entire + // category, not just the rows currently loaded. Optimistically flips the UI. + const markSectionReadAction = useCallback(async (section: InboxSection, read: boolean) => { + setSection(section, (prev) => ({ + ...prev, + threads: prev.threads.map((t) => ({ + ...t, + unread: !read, + messages: t.messages.map((m) => ({ ...m, unread: !read })), + })), + })) + try { + const result = await window.ipc.invoke('gmail:markSectionRead', { section, read }) + if (!result.ok && result.error) toast(`Could not update read state: ${result.error}`, 'error') + } catch (err) { + toast(`Could not update read state: ${err instanceof Error ? err.message : String(err)}`, 'error') + } + }, [setSection]) + const archiveThreadAction = useCallback(async (threadId: string) => { try { const result = await window.ipc.invoke('gmail:archiveThread', { threadId }) @@ -2151,6 +2446,9 @@ export function EmailView({ initialThreadId, threadIdVersion }: EmailViewProps = const visibleImportant = useMemo(() => filterThreads(important.threads), [important.threads, filterThreads]) const visibleOther = useMemo(() => filterThreads(other.threads), [other.threads, filterThreads]) + const visibleDrafts = useMemo(() => filterThreads(drafts), [drafts, filterThreads]) + const importantHasUnread = important.threads.some((t) => t.unread) + const otherHasUnread = other.threads.some((t) => t.unread) const hasAny = important.threads.length > 0 || other.threads.length > 0 const initialLoading = !hasAny && refreshing @@ -2186,17 +2484,15 @@ export function EmailView({ initialThreadId, threadIdVersion }: EmailViewProps = {formatInboxTime(latest?.date || thread.date)}
- {isUnread && ( - - )} + +
+ +
+
+ + ) + } + return (
@@ -2237,12 +2568,29 @@ export function EmailView({ initialThreadId, threadIdVersion }: EmailViewProps = setQuery(event.target.value)} - placeholder="Search loaded mail" + placeholder="Search all mail" />
- + +
+
- {error && !hasAny ? ( + {query.trim() ? ( + searchError && searchResults.length === 0 ? ( +
Could not search: {searchError}
+ ) : searchResults.length > 0 ? ( +
+
+
+ Search results + {searchResults.length} thread{searchResults.length === 1 ? '' : 's'} +
+ {searchResults.map(renderRow)} +
+
+ ) : ( +
+ {searching ? 'Searching all mail…' : `No results for “${query.trim()}”.`} +
+ ) + ) : view === 'drafts' ? ( + draftsError && drafts.length === 0 ? ( +
Could not load drafts: {draftsError}
+ ) : drafts.length > 0 ? ( +
+
+
+ Drafts + {drafts.length} draft{drafts.length === 1 ? '' : 's'} +
+ {visibleDrafts.map(renderDraftRow)} +
+
+ ) : ( +
+ {draftsLoading ? 'Loading drafts…' : 'No drafts yet.'} +
+ ) + ) : error && !hasAny ? (
Could not load mail: {error}
) : hasAny ? (
@@ -2258,9 +2642,19 @@ export function EmailView({ initialThreadId, threadIdVersion }: EmailViewProps =
Important - - {important.threads.length}{important.hasReachedEnd ? '' : '+'} thread{important.threads.length === 1 ? '' : 's'} - +
+
+ Read + void markSectionReadAction('important', checked)} + aria-label="Mark all of Important read or unread" + /> +
+ + {important.threads.length}{important.hasReachedEnd ? '' : '+'} thread{important.threads.length === 1 ? '' : 's'} + +
{visibleImportant.map(renderRow)} {!important.hasReachedEnd && ( @@ -2276,9 +2670,19 @@ export function EmailView({ initialThreadId, threadIdVersion }: EmailViewProps =
Everything else - - {other.threads.length}{other.hasReachedEnd ? '' : '+'} thread{other.threads.length === 1 ? '' : 's'} - +
+
+ Read + void markSectionReadAction('other', checked)} + aria-label="Mark all of Everything else read or unread" + /> +
+ + {other.threads.length}{other.hasReachedEnd ? '' : '+'} thread{other.threads.length === 1 ? '' : 's'} + +
{visibleOther.map(renderRow)} {!other.hasReachedEnd && ( @@ -2315,6 +2719,14 @@ export function EmailView({ initialThreadId, threadIdVersion }: EmailViewProps = )}
{composeOpen && } + {editingDraft && ( + + )} ) diff --git a/apps/x/packages/core/src/config/config.ts b/apps/x/packages/core/src/config/config.ts index 7f230514..82b2021e 100644 --- a/apps/x/packages/core/src/config/config.ts +++ b/apps/x/packages/core/src/config/config.ts @@ -44,6 +44,17 @@ function ensureDefaultConfigs() { configured: false }, null, 2)); } + + // Create gmail_sync.json with the default onboarding email count if it + // doesn't exist, so the "how many emails to backfill" setting is + // discoverable and editable. Keep the default in sync with + // DEFAULT_MAX_EMAILS in gmail_sync_config.ts. + const gmailSyncConfig = path.join(WorkDir, "config", "gmail_sync.json"); + if (!fs.existsSync(gmailSyncConfig)) { + fs.writeFileSync(gmailSyncConfig, JSON.stringify({ + maxEmails: 500 + }, null, 2)); + } } ensureDirs(); diff --git a/apps/x/packages/core/src/config/gmail_sync_config.ts b/apps/x/packages/core/src/config/gmail_sync_config.ts new file mode 100644 index 00000000..2a75f408 --- /dev/null +++ b/apps/x/packages/core/src/config/gmail_sync_config.ts @@ -0,0 +1,61 @@ +import fs from 'fs'; +import path from 'path'; +import { WorkDir } from './config.js'; + +const CONFIG_FILE = path.join(WorkDir, 'config', 'gmail_sync.json'); + +/** + * How many of the newest email threads the initial (onboarding) / recovery + * Gmail sync pulls down. This bounds the sync by a COUNT of recent threads + * rather than a fixed date window, so a fresh account backfills its most recent + * `maxEmails` emails even when they span more than a week. + */ +export const DEFAULT_MAX_EMAILS = 500; + +// Guard rails: at least one email, and a hard ceiling so a misconfigured value +// can't trigger a runaway onboarding sync (each thread costs a threads.get plus +// an LLM classification). +const MIN_MAX_EMAILS = 1; +const MAX_MAX_EMAILS = 5000; + +interface GmailSyncConfig { + maxEmails: number; +} + +function clampMaxEmails(value: number): number { + return Math.max(MIN_MAX_EMAILS, Math.min(MAX_MAX_EMAILS, Math.floor(value))); +} + +/** + * Read the configured max email count for the onboarding/full sync. + * Falls back to {@link DEFAULT_MAX_EMAILS} when the file is missing, malformed, + * or holds an out-of-range value. + */ +export function getMaxEmails(): number { + try { + if (fs.existsSync(CONFIG_FILE)) { + const raw = fs.readFileSync(CONFIG_FILE, 'utf-8'); + const parsed = JSON.parse(raw) as Partial; + const value = Number(parsed?.maxEmails); + if (Number.isFinite(value) && value > 0) { + return clampMaxEmails(value); + } + } + } catch (err) { + console.warn('[GmailSyncConfig] Failed to read gmail_sync.json:', err); + } + return DEFAULT_MAX_EMAILS; +} + +/** + * Persist the max email count used by the onboarding/full sync. The value is + * clamped into the supported range before writing. + */ +export function setMaxEmails(maxEmails: number): void { + const configDir = path.dirname(CONFIG_FILE); + if (!fs.existsSync(configDir)) { + fs.mkdirSync(configDir, { recursive: true }); + } + const config: GmailSyncConfig = { maxEmails: clampMaxEmails(maxEmails) }; + fs.writeFileSync(CONFIG_FILE, JSON.stringify(config, null, 2)); +} diff --git a/apps/x/packages/core/src/knowledge/sync_gmail.ts b/apps/x/packages/core/src/knowledge/sync_gmail.ts index e4d478e9..3270abd7 100644 --- a/apps/x/packages/core/src/knowledge/sync_gmail.ts +++ b/apps/x/packages/core/src/knowledge/sync_gmail.ts @@ -4,6 +4,7 @@ import { google, gmail_v1 as gmail } from 'googleapis'; import { NodeHtmlMarkdown } from 'node-html-markdown' import { OAuth2Client } from 'google-auth-library'; import { WorkDir } from '../config/config.js'; +import { getMaxEmails } from '../config/gmail_sync_config.js'; import { GoogleClientFactory } from './google-client-factory.js'; import { serviceLogger, type ServiceRunContext } from '../services/service_logger.js'; import { limitEventItems } from './limit_event_items.js'; @@ -15,6 +16,10 @@ import { notifyIfEnabled } from '../application/notification/notifier.js'; const SYNC_DIR = path.join(WorkDir, 'gmail_sync'); const LEGACY_CACHE_DIR = path.join(SYNC_DIR, 'cache'); const CACHE_DIR = path.join(WorkDir, 'inbox_lists'); +// Local index of full-text search results. Kept separate from inbox_lists/ so it +// never leaks non-inbox threads into the inbox view. Grows as you search; we +// don't prune it (the user wants a durable local index). +const SEARCH_CACHE_DIR = path.join(WorkDir, 'search_index'); (function migrateLegacyCacheDir() { try { @@ -95,6 +100,35 @@ function deleteCachedSnapshot(threadId: string): void { } } +// Local search index — same on-disk shape as the inbox cache, separate dir. +function searchCachePath(threadId: string): string { + return path.join(SEARCH_CACHE_DIR, `${encodeURIComponent(threadId)}.json`); +} + +function readSearchSnapshot(threadId: string): SnapshotCacheEntry | null { + try { + const raw = fs.readFileSync(searchCachePath(threadId), 'utf-8'); + return JSON.parse(raw) as SnapshotCacheEntry; + } catch { + return null; + } +} + +function writeSearchSnapshot(threadId: string, historyId: string, snapshot: GmailThreadSnapshot): void { + try { + if (!fs.existsSync(SEARCH_CACHE_DIR)) fs.mkdirSync(SEARCH_CACHE_DIR, { recursive: true }); + const entry: SnapshotCacheEntry = { + historyId, + fetchedAt: new Date().toISOString(), + parserVersion: SNAPSHOT_PARSER_VERSION, + snapshot, + }; + fs.writeFileSync(searchCachePath(threadId), JSON.stringify(entry), 'utf-8'); + } catch (err) { + console.warn(`[Gmail search index] write failed for ${threadId}:`, err); + } +} + async function getGmailClientOrThrow() { const auth = await GoogleClientFactory.getClient(); if (!auth) throw new Error('Gmail is not connected.'); @@ -132,19 +166,19 @@ export async function trashThread(threadId: string): Promise } } -export async function markThreadRead(threadId: string): Promise { +export async function markThreadRead(threadId: string, read: boolean = true): Promise { try { const gmailClient = await getGmailClientOrThrow(); await gmailClient.users.threads.modify({ userId: 'me', id: threadId, - requestBody: { removeLabelIds: ['UNREAD'] }, + requestBody: read ? { removeLabelIds: ['UNREAD'] } : { addLabelIds: ['UNREAD'] }, }); - // Update local cache: clear unread on all messages in the thread. + // Mirror the new read state onto every message in the cached thread. const cached = readCachedSnapshot(threadId); if (cached) { - for (const m of cached.snapshot.messages) m.unread = false; - cached.snapshot.unread = false; + for (const m of cached.snapshot.messages) m.unread = !read; + cached.snapshot.unread = !read; try { fs.writeFileSync(cachePath(threadId), JSON.stringify(cached), 'utf-8'); } catch (err) { @@ -157,6 +191,81 @@ export async function markThreadRead(threadId: string): Promise { + try { + const gmailClient = await getGmailClientOrThrow(); + + // Gather threads in this category that still need the change. + const targets: { file: string; entry: SnapshotCacheEntry; messageIds: string[] }[] = []; + if (fs.existsSync(CACHE_DIR)) { + let names: string[] = []; + try { names = fs.readdirSync(CACHE_DIR); } catch { names = []; } + for (const name of names) { + if (!name.endsWith('.json')) continue; + const file = path.join(CACHE_DIR, name); + let entry: SnapshotCacheEntry; + try { entry = JSON.parse(fs.readFileSync(file, 'utf-8')) as SnapshotCacheEntry; } catch { continue; } + const snap = entry?.snapshot; + if (!snap) continue; + const isUnread = snap.unread === true; + // Marking read: skip already-read. Marking unread: skip already-unread. + if (read ? !isUnread : isUnread) continue; + // Mirror snapshotImportance: missing importance counts as 'important'. + const importance = snap.importance === 'other' ? 'other' : 'important'; + if (importance !== section) continue; + const messageIds = snap.messages.map((m) => m.id).filter((id): id is string => Boolean(id)); + targets.push({ file, entry, messageIds }); + } + } + + if (targets.length === 0) return { ok: true, count: 0 }; + + // Apply on Gmail (batchModify caps at 1000 message ids per call). + const allMessageIds = targets.flatMap((t) => t.messageIds); + const labelChange = read ? { removeLabelIds: ['UNREAD'] } : { addLabelIds: ['UNREAD'] }; + for (let i = 0; i < allMessageIds.length; i += 1000) { + const ids = allMessageIds.slice(i, i + 1000); + if (ids.length === 0) continue; + await gmailClient.users.messages.batchModify({ + userId: 'me', + requestBody: { ids, ...labelChange }, + }); + } + + // Mirror into the local cache so the UI updates without a full resync. + for (const t of targets) { + for (const m of t.entry.snapshot.messages) m.unread = !read; + t.entry.snapshot.unread = !read; + try { + fs.writeFileSync(t.file, JSON.stringify(t.entry), 'utf-8'); + } catch (err) { + console.warn(`[Gmail cache] markSectionRead write failed:`, err); + } + } + + return { ok: true, count: targets.length }; + } catch (err) { + return { ok: false, count: 0, error: err instanceof Error ? err.message : String(err) }; + } +} + interface SyncedThread { threadId: string; markdown: string; @@ -176,6 +285,8 @@ export interface GmailThreadSnapshot { importance?: 'important' | 'other'; draft_response?: string; gmail_draft?: string; + /** Gmail-side draft id, present on entries from listDraftThreads. */ + draftId?: string; messages: Array<{ id?: string; from?: string; @@ -194,6 +305,7 @@ export interface GmailThreadSnapshot { savedPath: string; }>; messageIdHeader?: string; + isDraft?: boolean; }>; } @@ -792,6 +904,45 @@ async function buildAndCacheSnapshot( ) { return cached.snapshot; } + const snapshot = await parseThreadSnapshot(threadId, threadData, gmailClient); + if (!snapshot) return null; + + try { + const userEmail = await getUserEmail(auth); + const skipDraft = (snapshot.gmail_draft?.length ?? 0) > 0; + const classification = await classifyThread(snapshot, userEmail, { skipDraft }); + snapshot.importance = classification.importance; + if (classification.summary) snapshot.summary = classification.summary; + if (classification.draftResponse) { + const draftResponse = stripGmailQuotedReplyText(classification.draftResponse); + if (draftResponse) snapshot.draft_response = draftResponse; + } + } catch (err) { + console.warn(`[Gmail] classify failed for ${threadId}:`, err); + } + + if (threadData.historyId) { + writeCachedSnapshot(threadId, threadData.historyId, snapshot); + } + + return snapshot; +} + +/** + * Parse a threads.get response into a snapshot WITHOUT AI classification or + * caching — the shared core of buildAndCacheSnapshot, also used by search (which + * doesn't need importance/summary). Returns null when there are no visible + * (non-draft) messages. + */ +async function parseThreadSnapshot( + threadId: string, + threadData: gmail.Schema$Thread, + gmailClient: gmail.Gmail, +): Promise { + const messages = threadData.messages; + if (!messages || messages.length === 0) return null; + + const cached = readCachedSnapshot(threadId); const heightCarryover = new Map(); if (cached) { for (const m of cached.snapshot.messages) { @@ -855,7 +1006,7 @@ async function buildAndCacheSnapshot( .filter(Boolean) .join('\n\n'); - const snapshot: GmailThreadSnapshot = { + return { threadId, threadUrl: `https://mail.google.com/mail/u/0/#all/${threadId}`, subject: latest.subject || visibleMessages[0]?.subject, @@ -868,26 +1019,6 @@ async function buildAndCacheSnapshot( messages: visibleMessages, gmail_draft: latestDraftBody || undefined, }; - - try { - const userEmail = await getUserEmail(auth); - const skipDraft = latestDraftBody.length > 0; - const classification = await classifyThread(snapshot, userEmail, { skipDraft }); - snapshot.importance = classification.importance; - if (classification.summary) snapshot.summary = classification.summary; - if (classification.draftResponse) { - const draftResponse = stripGmailQuotedReplyText(classification.draftResponse); - if (draftResponse) snapshot.draft_response = draftResponse; - } - } catch (err) { - console.warn(`[Gmail] classify failed for ${threadId}:`, err); - } - - if (threadData.historyId) { - writeCachedSnapshot(threadId, threadData.historyId, snapshot); - } - - return snapshot; } async function saveAttachment(gmail: gmail.Gmail, userId: string, msgId: string, part: gmail.Schema$MessagePart, attachmentsDir: string): Promise { @@ -1148,24 +1279,29 @@ async function backfillMissingRecentThreads( async function fullSync(auth: OAuth2Client, syncDir: string, attachmentsDir: string, stateFile: string, lookbackDays: number) { const gmail = google.gmail({ version: 'v1', auth }); - // If the state file holds a last_sync timestamp (e.g. left over from a - // prior Composio sync, or from a previous successful native sync that - // we're falling back to after a history.list 404), use that as the - // floor — but never reach back further than lookbackDays. This caps the - // window at "1 week at most": if last_sync is within the lookback window - // we resume from it (a smaller window), otherwise we clamp to lookbackDays - // ago. Mail older than the cap that arrived during a long offline gap is - // intentionally skipped rather than backfilled. + // The onboarding / recovery fetch is bounded by a COUNT of the most recent + // threads (maxEmails, configurable — default 500), not by a fixed date + // window. So a fresh account pulls its newest `maxEmails` emails even when + // they span more than a week. + // + // When we can resume after a previous successful sync (a last_sync within + // the lookback window — e.g. the history.list 404 fallback, or a prior + // Composio sync), we still floor the query at last_sync so only genuinely + // new mail is re-walked, and the count cap acts purely as a safety bound. + // With no resumable last_sync (first connect, or a gap longer than the + // lookback window) we drop the date floor entirely and just take the newest + // `maxEmails` threads. + const maxEmails = getMaxEmails(); const state = loadState(stateFile); const lookbackFloor = new Date(); lookbackFloor.setDate(lookbackFloor.getDate() - lookbackDays); - let pastDate: Date; - if (state.last_sync && new Date(state.last_sync) > lookbackFloor) { - pastDate = new Date(state.last_sync); - console.log(`Performing full sync from last_sync=${state.last_sync}...`); + const resumeFrom = state.last_sync && new Date(state.last_sync) > lookbackFloor + ? new Date(state.last_sync) + : null; + if (resumeFrom) { + console.log(`Performing full sync from last_sync=${state.last_sync} (max ${maxEmails} threads)...`); } else { - pastDate = lookbackFloor; - console.log(`Performing full sync of last ${lookbackDays} days...`); + console.log(`Performing full sync of the newest ${maxEmails} threads...`); } let run: ServiceRunContext | null = null; @@ -1180,19 +1316,24 @@ async function fullSync(auth: OAuth2Client, syncDir: string, attachmentsDir: str }; try { - const dateQuery = pastDate.toISOString().split('T')[0].replace(/-/g, '/'); + const baseQuery = '-in:spam -in:trash'; + const q = resumeFrom + ? `after:${resumeFrom.toISOString().split('T')[0].replace(/-/g, '/')} ${baseQuery}` + : baseQuery; // Get History ID const profile = await gmail.users.getProfile({ userId: 'me' }); const currentHistoryId = profile.data.historyId!; + // Gmail returns threads newest-first, so paginating until we've collected + // maxEmails ids yields the most recent maxEmails threads. const threadIds: string[] = []; let pageToken: string | undefined; do { const res = await gmail.users.threads.list({ userId: 'me', - q: `after:${dateQuery} -in:spam -in:trash`, - maxResults: 500, + q, + maxResults: Math.min(500, maxEmails), pageToken }); @@ -1205,7 +1346,9 @@ async function fullSync(auth: OAuth2Client, syncDir: string, attachmentsDir: str } } pageToken = res.data.nextPageToken ?? undefined; - } while (pageToken); + } while (pageToken && threadIds.length < maxEmails); + + if (threadIds.length > maxEmails) threadIds.length = maxEmails; if (threadIds.length === 0) { saveState(currentHistoryId, stateFile); @@ -1434,10 +1577,10 @@ async function performSync() { // partial-sync on subsequent calls. const cacheMissing = !fs.existsSync(CACHE_DIR) || fs.readdirSync(CACHE_DIR).length === 0; // partialSync replays *every* messageAdded since the stored historyId, - // regardless of date — so after a long offline gap a still-valid - // historyId would pull the entire gap (e.g. 3 weeks). To honor the - // "1 week at most" cap, bypass it when last_sync is older than the - // lookback window and run a (date-clamped) fullSync instead. + // regardless of date/count — so after a long offline gap a still-valid + // historyId would pull the entire gap (e.g. 3 weeks). When last_sync is + // older than the lookback window, bypass it and run fullSync instead, + // which is count-bounded (the newest maxEmails threads). const gapMs = state.last_sync ? Date.now() - new Date(state.last_sync).getTime() : 0; const gapTooLarge = gapMs > LOOKBACK_DAYS * 24 * 60 * 60 * 1000; if (!state.historyId) { @@ -1447,7 +1590,7 @@ async function performSync() { console.log("History ID present but inbox cache empty — running full sync to backfill snapshots..."); await fullSync(auth, SYNC_DIR, ATTACHMENTS_DIR, STATE_FILE, LOOKBACK_DAYS); } else if (gapTooLarge) { - console.log(`Last sync older than ${LOOKBACK_DAYS} days — running full sync clamped to the lookback window instead of partial sync...`); + console.log(`Last sync older than ${LOOKBACK_DAYS} days — running count-bounded full sync instead of partial sync...`); await fullSync(auth, SYNC_DIR, ATTACHMENTS_DIR, STATE_FILE, LOOKBACK_DAYS); } else { console.log("History ID found, starting partial sync..."); @@ -1485,6 +1628,19 @@ export interface SendReplyResult { error?: string; } +export interface SaveDraftOptions extends Omit { + /** Recipient may be blank while a draft is still being written. */ + to?: string; + /** Existing Gmail draft to update; omitted on first save (creates a new one). */ + draftId?: string; +} + +export interface SaveDraftResult { + /** The Gmail-side draft id, to be passed back on subsequent saves. */ + draftId?: string; + error?: string; +} + export interface GmailConnectionStatus { connected: boolean; hasRequiredScope: boolean; @@ -1588,6 +1744,88 @@ function sanitizeAttachmentName(name: string): string { return (name || 'attachment').replace(/[\r\n"\\]/g, '_').trim() || 'attachment'; } +// Build the raw (base64url) RFC 2822 message shared by both send and draft-save. +// Recipient headers are omitted when blank, so an in-progress draft with no +// `To` yet still produces a valid message. `isEmpty` lets callers reject a +// whitespace-only body without re-parsing the result. +function buildRawMimeMessage(opts: SaveDraftOptions, userEmail: string): { raw: string; isEmpty: boolean } { + const safeTo = opts.to?.trim() ? requireSafeHeaderValue('To', opts.to) : undefined; + const safeCc = opts.cc?.trim() ? requireSafeHeaderValue('Cc', opts.cc) : undefined; + const safeBcc = opts.bcc?.trim() ? requireSafeHeaderValue('Bcc', opts.bcc) : undefined; + const safeInReplyTo = opts.inReplyTo ? requireSafeHeaderValue('In-Reply-To', opts.inReplyTo) : undefined; + const safeReferences = opts.references ? requireSafeHeaderValue('References', opts.references) : undefined; + const replyBody = opts.threadId + ? sanitizeReplyBodyForGmailReply(opts.bodyHtml, opts.bodyText) + : { bodyHtml: opts.bodyHtml.trim(), bodyText: opts.bodyText.trim() }; + const isEmpty = !replyBody.bodyText.trim(); + + const seed = `${Date.now()}_${Math.random().toString(36).slice(2, 10)}`; + const altBoundary = `alt_${seed}`; + const attachments = (opts.attachments ?? []).filter((a) => a.contentBase64); + + const headers: string[] = []; + headers.push(`From: ${requireSafeHeaderValue('From', userEmail)}`); + if (safeTo) headers.push(`To: ${safeTo}`); + if (safeCc) headers.push(`Cc: ${safeCc}`); + if (safeBcc) headers.push(`Bcc: ${safeBcc}`); + headers.push(`Subject: ${encodeRfc2047(opts.subject)}`); + if (safeInReplyTo) headers.push(`In-Reply-To: ${safeInReplyTo}`); + if (safeReferences) headers.push(`References: ${safeReferences}`); + headers.push('MIME-Version: 1.0'); + + // The text+html body as a self-contained multipart/alternative block. + const altParts: string[] = []; + altParts.push(`--${altBoundary}`); + altParts.push('Content-Type: text/plain; charset="UTF-8"'); + altParts.push('Content-Transfer-Encoding: base64'); + altParts.push(''); + altParts.push(encodeMimeBase64(replyBody.bodyText)); + altParts.push(''); + altParts.push(`--${altBoundary}`); + altParts.push('Content-Type: text/html; charset="UTF-8"'); + altParts.push('Content-Transfer-Encoding: base64'); + altParts.push(''); + altParts.push(encodeMimeBase64(replyBody.bodyHtml)); + altParts.push(''); + altParts.push(`--${altBoundary}--`); + + let body: string; + if (attachments.length) { + // Wrap the alternative body plus each attachment in a multipart/mixed. + const mixedBoundary = `mixed_${seed}`; + headers.push(`Content-Type: multipart/mixed; boundary="${mixedBoundary}"`); + const mixed: string[] = []; + mixed.push(`--${mixedBoundary}`); + mixed.push(`Content-Type: multipart/alternative; boundary="${altBoundary}"`); + mixed.push(''); + mixed.push(altParts.join('\r\n')); + for (const att of attachments) { + const name = sanitizeAttachmentName(att.filename); + const mime = sanitizeAttachmentName(att.mimeType) || 'application/octet-stream'; + mixed.push(`--${mixedBoundary}`); + mixed.push(`Content-Type: ${mime}; name="${name}"`); + mixed.push('Content-Transfer-Encoding: base64'); + mixed.push(`Content-Disposition: attachment; filename="${name}"`); + mixed.push(''); + mixed.push(wrapBase64(att.contentBase64)); + mixed.push(''); + } + mixed.push(`--${mixedBoundary}--`); + body = mixed.join('\r\n'); + } else { + headers.push(`Content-Type: multipart/alternative; boundary="${altBoundary}"`); + body = altParts.join('\r\n'); + } + + const message = `${headers.join('\r\n')}\r\n\r\n${body}`; + const raw = Buffer.from(message, 'utf8') + .toString('base64') + .replace(/\+/g, '-') + .replace(/\//g, '_') + .replace(/=+$/, ''); + return { raw, isEmpty }; +} + export async function sendThreadReply(opts: SendReplyOptions): Promise { try { const auth = await GoogleClientFactory.getClient(); @@ -1597,82 +1835,11 @@ export async function sendThreadReply(opts: SendReplyOptions): Promise a.contentBase64); - - const headers: string[] = []; - headers.push(`From: ${requireSafeHeaderValue('From', userEmail)}`); - headers.push(`To: ${safeTo}`); - if (safeCc) headers.push(`Cc: ${safeCc}`); - if (safeBcc) headers.push(`Bcc: ${safeBcc}`); - headers.push(`Subject: ${encodeRfc2047(opts.subject)}`); - if (safeInReplyTo) headers.push(`In-Reply-To: ${safeInReplyTo}`); - if (safeReferences) headers.push(`References: ${safeReferences}`); - headers.push('MIME-Version: 1.0'); - - // The text+html body as a self-contained multipart/alternative block. - const altParts: string[] = []; - altParts.push(`--${altBoundary}`); - altParts.push('Content-Type: text/plain; charset="UTF-8"'); - altParts.push('Content-Transfer-Encoding: base64'); - altParts.push(''); - altParts.push(encodeMimeBase64(replyBody.bodyText)); - altParts.push(''); - altParts.push(`--${altBoundary}`); - altParts.push('Content-Type: text/html; charset="UTF-8"'); - altParts.push('Content-Transfer-Encoding: base64'); - altParts.push(''); - altParts.push(encodeMimeBase64(replyBody.bodyHtml)); - altParts.push(''); - altParts.push(`--${altBoundary}--`); - - let body: string; - if (attachments.length) { - // Wrap the alternative body plus each attachment in a multipart/mixed. - const mixedBoundary = `mixed_${seed}`; - headers.push(`Content-Type: multipart/mixed; boundary="${mixedBoundary}"`); - const mixed: string[] = []; - mixed.push(`--${mixedBoundary}`); - mixed.push(`Content-Type: multipart/alternative; boundary="${altBoundary}"`); - mixed.push(''); - mixed.push(altParts.join('\r\n')); - for (const att of attachments) { - const name = sanitizeAttachmentName(att.filename); - const mime = sanitizeAttachmentName(att.mimeType) || 'application/octet-stream'; - mixed.push(`--${mixedBoundary}`); - mixed.push(`Content-Type: ${mime}; name="${name}"`); - mixed.push('Content-Transfer-Encoding: base64'); - mixed.push(`Content-Disposition: attachment; filename="${name}"`); - mixed.push(''); - mixed.push(wrapBase64(att.contentBase64)); - mixed.push(''); - } - mixed.push(`--${mixedBoundary}--`); - body = mixed.join('\r\n'); - } else { - headers.push(`Content-Type: multipart/alternative; boundary="${altBoundary}"`); - body = altParts.join('\r\n'); - } - - const message = `${headers.join('\r\n')}\r\n\r\n${body}`; - const raw = Buffer.from(message, 'utf8') - .toString('base64') - .replace(/\+/g, '-') - .replace(/\//g, '_') - .replace(/=+$/, ''); - - const requestBody: gmail.Schema$Message = { raw }; + const requestBody: gmail.Schema$Message = { raw: built.raw }; if (opts.threadId) requestBody.threadId = opts.threadId; const res = await gmailClient.users.messages.send({ @@ -1706,6 +1873,265 @@ export async function sendThreadReply(opts: SendReplyOptions): Promise { + try { + const auth = await GoogleClientFactory.getClient(); + if (!auth) return { error: 'Gmail is not connected.' }; + + const gmailClient = google.gmail({ version: 'v1', auth }); + const userEmail = await getUserEmail(auth); + if (!userEmail) return { error: 'Could not determine your Gmail address.' }; + + const built = buildRawMimeMessage(opts, userEmail); + if (built.isEmpty) return { error: 'Draft is empty.' }; + + const message: gmail.Schema$Message = { raw: built.raw }; + if (opts.threadId) message.threadId = opts.threadId; + + // Resolve which draft to update: explicit id wins; otherwise reuse an + // existing draft on the same thread so replies don't pile up duplicates. + let draftId = opts.draftId; + if (!draftId && opts.threadId) { + try { + const drafts = await gmailClient.users.drafts.list({ userId: 'me' }); + const existing = (drafts.data.drafts || []).find( + (d) => d.message?.threadId === opts.threadId && d.id + ); + if (existing?.id) draftId = existing.id; + } catch { + // Listing failed — fall through and create a new draft. + } + } + + let res; + if (draftId) { + try { + res = await gmailClient.users.drafts.update({ + userId: 'me', + id: draftId, + requestBody: { message }, + }); + } catch { + // The draft was likely deleted or already sent; start a new one. + res = await gmailClient.users.drafts.create({ userId: 'me', requestBody: { message } }); + } + } else { + res = await gmailClient.users.drafts.create({ userId: 'me', requestBody: { message } }); + } + + // Wake the sync loop so the local cache reflects the draft. + triggerSync(); + + return { draftId: res.data.id || undefined }; + } catch (err) { + return { error: err instanceof Error ? err.message : String(err) }; + } +} + +/** Delete a Gmail draft by id. A missing draft is treated as success. */ +export async function deleteThreadDraft(draftId: string): Promise<{ ok: boolean; error?: string }> { + try { + const auth = await GoogleClientFactory.getClient(); + if (!auth) return { ok: false, error: 'Gmail is not connected.' }; + + const gmailClient = google.gmail({ version: 'v1', auth }); + await gmailClient.users.drafts.delete({ userId: 'me', id: draftId }); + triggerSync(); + return { ok: true }; + } catch (err) { + const code = (err as { code?: number; response?: { status?: number } })?.code + ?? (err as { response?: { status?: number } })?.response?.status; + // Already gone (sent/deleted) — nothing to do. + if (code === 404 || code === 410) return { ok: true }; + return { ok: false, error: err instanceof Error ? err.message : String(err) }; + } +} + +// In-memory cache of built draft snapshots, keyed by draftId and validated by +// the draft's underlying message id. Gmail assigns a fresh message id whenever a +// draft is updated (locally via saveThreadDraft or in another client), so an +// unchanged message id means the parsed snapshot can be reused — we skip the +// per-draft drafts.get + body parse, mirroring listInboxPage's mtime cache. +interface DraftCacheEntry { + messageId: string; + snapshot: GmailThreadSnapshot; +} +const draftListCache = new Map(); + +// Fetch one draft and parse it into a lightweight snapshot for the Drafts view. +async function buildDraftSnapshot( + gmailClient: gmail.Gmail, + draftId: string, +): Promise { + const full = await gmailClient.users.drafts.get({ userId: 'me', id: draftId, format: 'full' }); + const msg = full.data.message; + if (!msg) return null; + + const headers = msg.payload?.headers || []; + const parts = msg.payload ? extractBodyParts(msg.payload) : { text: '', html: '' }; + const rawBody = msg.payload ? normalizeBody(getBody(msg.payload)) : ''; + const body = stripGmailQuotedReplyText(rawBody); + const subject = headerValue(headers, 'Subject') || ''; + const from = headerValue(headers, 'From') || ''; + const to = headerValue(headers, 'To') || ''; + const cc = headerValue(headers, 'Cc') || ''; + const date = headerValue(headers, 'Date') || ''; + const threadId = msg.threadId || draftId; + const messageIdHeader = + headerValue(headers, 'Message-ID') || headerValue(headers, 'Message-Id') || undefined; + + return { + threadId, + threadUrl: `https://mail.google.com/mail/u/0/#drafts?compose=${draftId}`, + subject, + from, + to, + date, + latest_email: body, + gmail_draft: body || undefined, + draftId, + unread: false, + messages: [{ + id: msg.id || undefined, + from, + to, + cc: cc || undefined, + date, + subject, + body, + bodyHtml: parts.html || undefined, + messageIdHeader, + isDraft: true, + }], + }; +} + +/** + * List the account's Gmail drafts (reply drafts and standalone new-message + * drafts) as lightweight thread snapshots for the Drafts view. Drafts aren't + * part of the INBOX snapshot cache, so we read them from the Gmail API — but a + * cheap drafts.list (ids only) lets us reuse already-parsed snapshots for + * unchanged drafts and only drafts.get the new/edited ones. No AI + * classification; recipients/subject/body come straight off the draft message. + */ +export async function listDraftThreads(): Promise<{ threads: GmailThreadSnapshot[]; error?: string }> { + try { + const auth = await GoogleClientFactory.getClient(); + if (!auth) { + draftListCache.clear(); + return { threads: [], error: 'Gmail is not connected.' }; + } + + const gmailClient = google.gmail({ version: 'v1', auth }); + const list = await gmailClient.users.drafts.list({ userId: 'me', maxResults: 50 }); + const drafts = list.data.drafts || []; + + const seen = new Set(); + const built = await Promise.all(drafts.map(async (d) => { + if (!d.id) return null; + seen.add(d.id); + const messageId = d.message?.id || ''; + const cached = draftListCache.get(d.id); + // Reuse the cached snapshot when the draft's message id is unchanged. + if (cached && messageId && cached.messageId === messageId) { + return cached.snapshot; + } + try { + const snapshot = await buildDraftSnapshot(gmailClient, d.id); + if (snapshot) draftListCache.set(d.id, { messageId, snapshot }); + return snapshot; + } catch (err) { + console.warn('[Gmail] draft fetch failed:', err); + // Fall back to a stale cached copy if we have one. + return cached?.snapshot ?? null; + } + })); + + // Evict cache entries for drafts that no longer exist (sent/deleted). + for (const key of draftListCache.keys()) { + if (!seen.has(key)) draftListCache.delete(key); + } + + const threads = built.filter((s): s is GmailThreadSnapshot => s !== null); + // Newest first. + threads.sort((a, b) => { + const da = a.date ? Date.parse(a.date) : 0; + const db = b.date ? Date.parse(b.date) : 0; + return (Number.isFinite(db) ? db : 0) - (Number.isFinite(da) ? da : 0); + }); + return { threads }; + } catch (err) { + return { threads: [], error: err instanceof Error ? err.message : String(err) }; + } +} + +export interface SearchResult { + threads: GmailThreadSnapshot[]; + error?: string; +} + +/** + * Full-text search across the ENTIRE Gmail mailbox (not just locally-synced + * mail) using Gmail's `q` query. Each matching thread is parsed into a snapshot + * and written to the local search index so repeat searches — and opening a + * result — are instant. Reuses the inbox cache when a thread is already synced + * there. No AI classification. + */ +export async function searchThreads(query: string, opts: { limit?: number } = {}): Promise { + const q = query.trim(); + if (!q) return { threads: [] }; + try { + const auth = await GoogleClientFactory.getClient(); + if (!auth) return { threads: [], error: 'Gmail is not connected.' }; + + const gmailClient = google.gmail({ version: 'v1', auth }); + // Generous cap so the index isn't artificially small (Gmail allows 500). + const limit = Math.max(1, Math.min(200, opts.limit ?? 100)); + const list = await gmailClient.users.threads.list({ userId: 'me', q, maxResults: limit }); + const ids = (list.data.threads || []) + .map((t) => t.id) + .filter((id): id is string => Boolean(id)); + + const built = await Promise.all(ids.map(async (threadId) => { + // Prefer the inbox snapshot (kept fresh by sync), then the search index. + const inboxCached = readCachedSnapshot(threadId); + if (inboxCached?.snapshot) return inboxCached.snapshot; + const indexed = readSearchSnapshot(threadId); + if (indexed?.snapshot) return indexed.snapshot; + try { + const threadData = await gmailClient.users.threads.get({ userId: 'me', id: threadId, format: 'full' }); + const snapshot = await parseThreadSnapshot(threadId, threadData.data, gmailClient); + if (snapshot) writeSearchSnapshot(threadId, threadData.data.historyId || '', snapshot); + return snapshot; + } catch (err) { + console.warn(`[Gmail search] fetch failed for ${threadId}:`, err); + return null; + } + })); + + const threads = built.filter((s): s is GmailThreadSnapshot => s !== null); + // Newest first. + threads.sort((a, b) => { + const da = a.date ? Date.parse(a.date) : 0; + const db = b.date ? Date.parse(b.date) : 0; + return (Number.isFinite(db) ? db : 0) - (Number.isFinite(da) ? da : 0); + }); + return { threads }; + } catch (err) { + return { threads: [], error: err instanceof Error ? err.message : String(err) }; + } +} + export async function init() { console.log("Starting Gmail Sync (TS)..."); console.log(`Will sync every ${SYNC_INTERVAL_MS / 1000} seconds.`); diff --git a/apps/x/packages/shared/src/blocks.ts b/apps/x/packages/shared/src/blocks.ts index 69b331c4..e65e2b5f 100644 --- a/apps/x/packages/shared/src/blocks.ts +++ b/apps/x/packages/shared/src/blocks.ts @@ -124,6 +124,8 @@ export const GmailThreadMessageSchema = z.object({ bodyHeight: z.number().int().positive().optional(), attachments: z.array(GmailAttachmentSchema).optional(), messageIdHeader: z.string().optional(), + // Set on the unsent draft message within a thread (used by the Drafts view). + isDraft: z.boolean().optional(), }); export type GmailThreadMessage = z.infer; @@ -134,6 +136,9 @@ export const GmailThreadSchema = EmailBlockSchema.extend({ unread: z.boolean().optional(), importance: z.enum(['important', 'other']).optional(), gmail_draft: z.string().optional(), + // Gmail-side draft id, present on entries returned by the Drafts list so the + // composer can update/delete that exact draft. + draftId: z.string().optional(), messages: z.array(GmailThreadMessageSchema), }); diff --git a/apps/x/packages/shared/src/ipc.ts b/apps/x/packages/shared/src/ipc.ts index d1543006..79583141 100644 --- a/apps/x/packages/shared/src/ipc.ts +++ b/apps/x/packages/shared/src/ipc.ts @@ -205,6 +205,56 @@ const ipcSchemas = { error: z.string().optional(), }), }, + 'gmail:saveDraft': { + req: z.object({ + // Existing Gmail draft to update; omitted on first save (creates a new one). + draftId: z.string().min(1).optional(), + threadId: z.string().min(1).optional(), + // Recipients may be blank for a draft (unlike a send). + to: z.string().optional(), + cc: z.string().optional(), + bcc: z.string().optional(), + subject: z.string(), + bodyHtml: z.string(), + bodyText: z.string(), + inReplyTo: z.string().optional(), + references: z.string().optional(), + attachments: z + .array( + z.object({ + filename: z.string(), + mimeType: z.string(), + contentBase64: z.string(), + }), + ) + .optional(), + }), + res: z.object({ + draftId: z.string().optional(), + error: z.string().optional(), + }), + }, + 'gmail:deleteDraft': { + req: z.object({ draftId: z.string().min(1) }), + res: z.object({ ok: z.boolean(), error: z.string().optional() }), + }, + 'gmail:getDrafts': { + req: z.object({}), + res: z.object({ + threads: z.array(GmailThreadSchema), + error: z.string().optional(), + }), + }, + 'gmail:search': { + req: z.object({ + query: z.string(), + limit: z.number().int().positive().optional(), + }), + res: z.object({ + threads: z.array(GmailThreadSchema), + error: z.string().optional(), + }), + }, 'gmail:getConnectionStatus': { req: z.object({}), res: z.object({ @@ -235,9 +285,13 @@ const ipcSchemas = { res: z.object({ ok: z.boolean(), error: z.string().optional() }), }, 'gmail:markThreadRead': { - req: z.object({ threadId: z.string().min(1) }), + req: z.object({ threadId: z.string().min(1), read: z.boolean().optional() }), res: z.object({ ok: z.boolean(), error: z.string().optional() }), }, + 'gmail:markSectionRead': { + req: z.object({ section: z.enum(['important', 'other']), read: z.boolean() }), + res: z.object({ ok: z.boolean(), count: z.number().int().nonnegative(), error: z.string().optional() }), + }, 'gmail:saveMessageHeight': { req: z.object({ threadId: z.string().min(1),