diff --git a/src/app/(app)/settings/_components/EmailAccountsSection.tsx b/src/app/(app)/settings/_components/EmailAccountsSection.tsx index afc49346..ded6ae80 100644 --- a/src/app/(app)/settings/_components/EmailAccountsSection.tsx +++ b/src/app/(app)/settings/_components/EmailAccountsSection.tsx @@ -95,50 +95,48 @@ export function EmailAccountsSection({ {initial.length === 0 ? (
-
- - - - - - -

- Connect Your Gmail + Connect Your Inbox

- Read and send email from your Gmail inbox without leaving the portal. + Read and send email from the portal using your existing inbox.

- - Connect Gmail - +
+ + Connect Gmail + + + Connect Outlook + +
) : ( - initial.map((acct) => ( - - )) + <> + {initial.map((acct) => ( + + ))} +
+ + Connect Another Gmail + + · + + Connect Outlook + +
+ )} ); diff --git a/src/app/api/oauth/microsoft/callback/route.ts b/src/app/api/oauth/microsoft/callback/route.ts new file mode 100644 index 00000000..bb4530c2 --- /dev/null +++ b/src/app/api/oauth/microsoft/callback/route.ts @@ -0,0 +1,119 @@ +import { NextRequest, NextResponse } from "next/server"; +import { createClient } from "@/lib/supabase/server"; +import { createServiceClient } from "@/lib/supabase/service"; +import { + exchangeCodeForTokens, + getUserInfo, +} from "@/lib/email/microsoft-oauth"; +import { encryptToken } from "@/lib/email/crypto"; +import { syncOutlookAccount } from "@/lib/email/sync"; + +export const dynamic = "force-dynamic"; +export const maxDuration = 300; + +export async function GET(req: NextRequest) { + const url = req.nextUrl; + const code = url.searchParams.get("code"); + const stateFromQuery = url.searchParams.get("state"); + const error = url.searchParams.get("error"); + const stateCookie = req.cookies.get("microsoft_oauth_state")?.value; + + if (error) { + return NextResponse.redirect( + new URL(`/settings?email_connect=error&reason=${error}`, req.url) + ); + } + if ( + !code || + !stateFromQuery || + !stateCookie || + stateFromQuery !== stateCookie + ) { + return NextResponse.redirect( + new URL("/settings?email_connect=error&reason=state_mismatch", req.url) + ); + } + + const sb = await createClient(); + const { + data: { user }, + } = await sb.auth.getUser(); + if (!user) { + return NextResponse.redirect(new URL("/login", req.url)); + } + + const { data: profile } = await sb + .from("profiles") + .select("org_id") + .eq("id", user.id) + .maybeSingle(); + if (!profile?.org_id) { + return NextResponse.redirect( + new URL("/settings?email_connect=error&reason=no_org", req.url) + ); + } + + let tokens, userInfo; + try { + tokens = await exchangeCodeForTokens({ code, origin: url.origin }); + userInfo = await getUserInfo(tokens.access_token); + } catch (e) { + console.error("MS OAuth exchange failed", e); + return NextResponse.redirect( + new URL( + "/settings?email_connect=error&reason=exchange_failed", + req.url + ) + ); + } + + const svc = createServiceClient(); + const expiresAt = new Date( + Date.now() + tokens.expires_in * 1000 + ).toISOString(); + + const { error: upsertErr } = await svc.from("channel_accounts").upsert( + { + org_id: profile.org_id, + user_id: user.id, + provider: "outlook", + address: userInfo.email, + display_name: userInfo.name ?? null, + access_token_encrypted: encryptToken(tokens.access_token), + refresh_token_encrypted: tokens.refresh_token + ? encryptToken(tokens.refresh_token) + : null, + token_expires_at: expiresAt, + status: "active", + }, + { onConflict: "user_id,provider,address" } + ); + + if (upsertErr) { + console.error("channel_accounts upsert failed", upsertErr); + return NextResponse.redirect( + new URL("/settings?email_connect=error&reason=db_write", req.url) + ); + } + + const { data: justConnected } = await svc + .from("channel_accounts") + .select("id") + .eq("user_id", user.id) + .eq("provider", "outlook") + .eq("address", userInfo.email) + .maybeSingle(); + if (justConnected?.id) { + try { + await syncOutlookAccount(justConnected.id as string); + } catch (e) { + console.error("initial outlook sync failed", e); + } + } + + const redirect = NextResponse.redirect( + new URL("/settings?email_connect=success#email-accounts", req.url) + ); + redirect.cookies.delete("microsoft_oauth_state"); + return redirect; +} diff --git a/src/app/api/oauth/microsoft/start/route.ts b/src/app/api/oauth/microsoft/start/route.ts new file mode 100644 index 00000000..0a88e432 --- /dev/null +++ b/src/app/api/oauth/microsoft/start/route.ts @@ -0,0 +1,29 @@ +import { NextRequest, NextResponse } from "next/server"; +import { randomBytes } from "node:crypto"; +import { createClient } from "@/lib/supabase/server"; +import { buildAuthorizeUrl } from "@/lib/email/microsoft-oauth"; + +export const dynamic = "force-dynamic"; + +export async function GET(req: NextRequest) { + const sb = await createClient(); + const { + data: { user }, + } = await sb.auth.getUser(); + if (!user) { + return NextResponse.redirect(new URL("/login", req.url)); + } + + const state = randomBytes(24).toString("hex"); + const origin = req.nextUrl.origin; + + const res = NextResponse.redirect(buildAuthorizeUrl({ origin, state })); + res.cookies.set("microsoft_oauth_state", state, { + httpOnly: true, + secure: process.env.NODE_ENV === "production", + sameSite: "lax", + maxAge: 600, + path: "/", + }); + return res; +} diff --git a/src/lib/email/microsoft-oauth.ts b/src/lib/email/microsoft-oauth.ts new file mode 100644 index 00000000..845a67a3 --- /dev/null +++ b/src/lib/email/microsoft-oauth.ts @@ -0,0 +1,121 @@ +const OUTLOOK_SCOPES = [ + "offline_access", + "openid", + "email", + "profile", + "User.Read", + "Mail.Read", + "Mail.Send", + "Mail.ReadWrite", +]; + +const TENANT = "common"; + +export function getRedirectUri(origin: string): string { + return `${origin}/api/oauth/microsoft/callback`; +} + +export function buildAuthorizeUrl(opts: { + origin: string; + state: string; +}): string { + const clientId = process.env.MICROSOFT_CLIENT_ID; + if (!clientId) throw new Error("MICROSOFT_CLIENT_ID is not set."); + + const params = new URLSearchParams({ + client_id: clientId, + redirect_uri: getRedirectUri(opts.origin), + response_type: "code", + response_mode: "query", + scope: OUTLOOK_SCOPES.join(" "), + state: opts.state, + prompt: "consent", + }); + return `https://login.microsoftonline.com/${TENANT}/oauth2/v2.0/authorize?${params.toString()}`; +} + +export type MicrosoftTokenResponse = { + access_token: string; + refresh_token?: string; + expires_in: number; + scope: string; + token_type: string; + id_token?: string; +}; + +export async function exchangeCodeForTokens(opts: { + code: string; + origin: string; +}): Promise { + const clientId = process.env.MICROSOFT_CLIENT_ID; + const clientSecret = process.env.MICROSOFT_CLIENT_SECRET; + if (!clientId || !clientSecret) { + throw new Error("MICROSOFT_CLIENT_ID/SECRET not set."); + } + const res = await fetch( + `https://login.microsoftonline.com/${TENANT}/oauth2/v2.0/token`, + { + method: "POST", + headers: { "Content-Type": "application/x-www-form-urlencoded" }, + body: new URLSearchParams({ + code: opts.code, + client_id: clientId, + client_secret: clientSecret, + redirect_uri: getRedirectUri(opts.origin), + grant_type: "authorization_code", + scope: OUTLOOK_SCOPES.join(" "), + }), + } + ); + if (!res.ok) { + throw new Error(`Token exchange failed: ${res.status} ${await res.text()}`); + } + return (await res.json()) as MicrosoftTokenResponse; +} + +export async function refreshAccessToken(opts: { + refreshToken: string; +}): Promise { + const clientId = process.env.MICROSOFT_CLIENT_ID; + const clientSecret = process.env.MICROSOFT_CLIENT_SECRET; + if (!clientId || !clientSecret) { + throw new Error("MICROSOFT_CLIENT_ID/SECRET not set."); + } + const res = await fetch( + `https://login.microsoftonline.com/${TENANT}/oauth2/v2.0/token`, + { + method: "POST", + headers: { "Content-Type": "application/x-www-form-urlencoded" }, + body: new URLSearchParams({ + client_id: clientId, + client_secret: clientSecret, + refresh_token: opts.refreshToken, + grant_type: "refresh_token", + scope: OUTLOOK_SCOPES.join(" "), + }), + } + ); + if (!res.ok) { + throw new Error(`Token refresh failed: ${res.status} ${await res.text()}`); + } + return (await res.json()) as MicrosoftTokenResponse; +} + +export async function getUserInfo( + accessToken: string +): Promise<{ email: string; name?: string }> { + const res = await fetch("https://graph.microsoft.com/v1.0/me", { + headers: { Authorization: `Bearer ${accessToken}` }, + }); + if (!res.ok) { + throw new Error(`Graph /me failed: ${res.status} ${await res.text()}`); + } + const body = (await res.json()) as { + mail?: string | null; + userPrincipalName?: string | null; + displayName?: string | null; + }; + const email = (body.mail ?? body.userPrincipalName ?? "").toLowerCase(); + if (!email) throw new Error("Graph /me returned no email address"); + return { email, name: body.displayName ?? undefined }; +} diff --git a/src/lib/email/outlook.ts b/src/lib/email/outlook.ts new file mode 100644 index 00000000..f004ecbf --- /dev/null +++ b/src/lib/email/outlook.ts @@ -0,0 +1,231 @@ +import { refreshAccessToken } from "./microsoft-oauth"; +import { encryptToken, decryptToken } from "./crypto"; +import { createServiceClient } from "../supabase/service"; + +type AccountRow = { + id: string; + access_token_encrypted: string | null; + refresh_token_encrypted: string | null; + token_expires_at: string | null; +}; + +const GRAPH_BASE = "https://graph.microsoft.com/v1.0"; + +async function loadAccount(accountId: string): Promise { + const sb = createServiceClient(); + const { data, error } = await sb + .from("channel_accounts") + .select( + "id, access_token_encrypted, refresh_token_encrypted, token_expires_at" + ) + .eq("id", accountId) + .maybeSingle(); + if (error || !data) { + throw new Error(`channel_account ${accountId} not found`); + } + return data as AccountRow; +} + +async function persistRefreshedTokens(opts: { + accountId: string; + accessToken: string; + refreshToken: string | null; + expiresIn: number; +}) { + const sb = createServiceClient(); + const patch: Record = { + access_token_encrypted: encryptToken(opts.accessToken), + token_expires_at: new Date( + Date.now() + opts.expiresIn * 1000 + ).toISOString(), + }; + if (opts.refreshToken) { + patch.refresh_token_encrypted = encryptToken(opts.refreshToken); + } + await sb.from("channel_accounts").update(patch).eq("id", opts.accountId); +} + +async function getValidAccessToken(accountId: string): Promise { + const account = await loadAccount(accountId); + if (!account.access_token_encrypted || !account.refresh_token_encrypted) { + throw new Error(`channel_account ${accountId} missing tokens`); + } + const expiresAt = account.token_expires_at + ? new Date(account.token_expires_at).getTime() + : 0; + if (expiresAt - Date.now() > 120_000) { + return decryptToken(account.access_token_encrypted); + } + const refreshed = await refreshAccessToken({ + refreshToken: decryptToken(account.refresh_token_encrypted), + }); + await persistRefreshedTokens({ + accountId, + accessToken: refreshed.access_token, + refreshToken: refreshed.refresh_token ?? null, + expiresIn: refreshed.expires_in, + }); + return refreshed.access_token; +} + +async function graphFetch( + accountId: string, + pathOrUrl: string, + init: RequestInit = {} +): Promise { + const url = pathOrUrl.startsWith("http") + ? pathOrUrl + : `${GRAPH_BASE}${pathOrUrl}`; + let token = await getValidAccessToken(accountId); + let res = await fetch(url, { + ...init, + headers: { + ...(init.headers ?? {}), + Authorization: `Bearer ${token}`, + }, + }); + if (res.status === 401) { + const account = await loadAccount(accountId); + if (!account.refresh_token_encrypted) { + throw new Error(`channel_account ${accountId} missing refresh token`); + } + const refreshed = await refreshAccessToken({ + refreshToken: decryptToken(account.refresh_token_encrypted), + }); + await persistRefreshedTokens({ + accountId, + accessToken: refreshed.access_token, + refreshToken: refreshed.refresh_token ?? null, + expiresIn: refreshed.expires_in, + }); + token = refreshed.access_token; + res = await fetch(url, { + ...init, + headers: { + ...(init.headers ?? {}), + Authorization: `Bearer ${token}`, + }, + }); + } + if (!res.ok) { + throw new Error(`Graph ${res.status}: ${await res.text()}`); + } + return (await res.json()) as T; +} + +export type OutlookEmailAddress = { + emailAddress: { name?: string; address: string }; +}; + +export type OutlookAttachmentMeta = { + id: string; + name: string; + contentType: string; + size: number; + isInline: boolean; +}; + +export type OutlookMessage = { + id: string; + conversationId: string; + internetMessageId?: string; + subject?: string; + bodyPreview?: string; + body?: { contentType: "html" | "text"; content: string }; + from?: OutlookEmailAddress; + toRecipients?: OutlookEmailAddress[]; + ccRecipients?: OutlookEmailAddress[]; + bccRecipients?: OutlookEmailAddress[]; + receivedDateTime?: string; + sentDateTime?: string; + isRead?: boolean; + hasAttachments?: boolean; + internetMessageHeaders?: { name: string; value: string }[]; +}; + +export type DeltaPage = { + value: OutlookMessage[]; + "@odata.nextLink"?: string; + "@odata.deltaLink"?: string; +}; + +const SELECT_FIELDS = [ + "id", + "conversationId", + "internetMessageId", + "subject", + "bodyPreview", + "body", + "from", + "toRecipients", + "ccRecipients", + "bccRecipients", + "receivedDateTime", + "sentDateTime", + "isRead", + "hasAttachments", + "internetMessageHeaders", +].join(","); + +export async function listInboxDelta(opts: { + accountId: string; + deltaLinkOrInitial: string | null; + pageUrl?: string; +}): Promise { + if (opts.pageUrl) { + return graphFetch(opts.accountId, opts.pageUrl); + } + if (opts.deltaLinkOrInitial) { + return graphFetch(opts.accountId, opts.deltaLinkOrInitial); + } + return graphFetch( + opts.accountId, + `/me/mailFolders('inbox')/messages/delta?$select=${SELECT_FIELDS}&$top=50` + ); +} + +export async function listAttachments(opts: { + accountId: string; + messageId: string; +}): Promise<{ value: OutlookAttachmentMeta[] }> { + return graphFetch( + opts.accountId, + `/me/messages/${opts.messageId}/attachments?$select=id,name,contentType,size,isInline` + ); +} + +export async function getAttachmentContent(opts: { + accountId: string; + messageId: string; + attachmentId: string; +}): Promise { + const token = await getValidAccessToken(opts.accountId); + const res = await fetch( + `${GRAPH_BASE}/me/messages/${opts.messageId}/attachments/${opts.attachmentId}/$value`, + { headers: { Authorization: `Bearer ${token}` } } + ); + if (!res.ok) { + throw new Error(`Attachment fetch ${res.status}: ${await res.text()}`); + } + return Buffer.from(await res.arrayBuffer()); +} + +export function addressListOf( + recips: OutlookEmailAddress[] | undefined +): string[] { + return (recips ?? []) + .map((r) => r.emailAddress?.address ?? "") + .filter(Boolean) + .map((a) => a.toLowerCase()); +} + +export function getHeader( + msg: OutlookMessage, + name: string +): string | null { + const lower = name.toLowerCase(); + for (const h of msg.internetMessageHeaders ?? []) { + if (h.name.toLowerCase() === lower) return h.value; + } + return null; +} diff --git a/src/lib/email/sync.ts b/src/lib/email/sync.ts index d2492884..f2ee888d 100644 --- a/src/lib/email/sync.ts +++ b/src/lib/email/sync.ts @@ -13,6 +13,14 @@ import { parseFromHeader, type GmailMessageFull, } from "./gmail"; +import { + listInboxDelta, + listAttachments as listOutlookAttachments, + getAttachmentContent as getOutlookAttachmentContent, + addressListOf, + getHeader as getOutlookHeader, + type OutlookMessage, +} from "./outlook"; import { findLeadForAddress } from "./auto-link"; const BACKFILL_QUERY = "newer_than:90d"; @@ -417,18 +425,356 @@ async function upsertConversation(opts: { }; } -// Convenience used by the cron route — syncs every active gmail account. +export async function syncOutlookAccount( + accountId: string +): Promise { + const svc = createServiceClient(); + const result: SyncResult = { + accountId, + mode: "backfill", + messagesIngested: 0, + attachmentsStored: 0, + errors: [], + }; + + const { data: acct, error: acctErr } = await svc + .from("channel_accounts") + .select("id, org_id, user_id, address, sync_cursor, provider") + .eq("id", accountId) + .maybeSingle(); + if (acctErr || !acct) throw new Error(`account ${accountId} not found`); + if (acct.provider !== "outlook") return result; + + result.mode = acct.sync_cursor ? "incremental" : "backfill"; + + let pageUrl: string | undefined; + let deltaLink: string | null = null; + let pages = 0; + const MAX_PAGES = 25; + + try { + while (pages < MAX_PAGES) { + const page = await listInboxDelta({ + accountId, + deltaLinkOrInitial: pageUrl ? null : acct.sync_cursor, + pageUrl, + }); + for (const msg of page.value ?? []) { + try { + const ingested = await ingestOutlookMessage(accountId, acct, msg); + if (ingested) result.messagesIngested += 1; + result.attachmentsStored += ingested?.attachments ?? 0; + } catch (e) { + result.errors.push( + `msg ${msg.id}: ${e instanceof Error ? e.message : e}` + ); + } + } + pages += 1; + if (page["@odata.nextLink"]) { + pageUrl = page["@odata.nextLink"]; + continue; + } + if (page["@odata.deltaLink"]) { + deltaLink = page["@odata.deltaLink"]; + } + break; + } + } catch (e) { + const msg = e instanceof Error ? e.message : String(e); + // Graph returns 410 Gone when the delta token is too old; retry with a + // fresh delta query. + if (msg.includes("410") && acct.sync_cursor) { + await svc + .from("channel_accounts") + .update({ sync_cursor: null }) + .eq("id", accountId); + result.errors.push("delta token expired, will full-resync next run"); + } else { + result.errors.push(msg); + } + } + + if (deltaLink) { + await svc + .from("channel_accounts") + .update({ sync_cursor: deltaLink }) + .eq("id", accountId); + } + + await svc + .from("channel_accounts") + .update({ last_synced_at: new Date().toISOString() }) + .eq("id", accountId); + + return result; +} + +async function ingestOutlookMessage( + accountId: string, + acct: AccountCtx, + msg: OutlookMessage +): Promise<{ attachments: number } | null> { + const svc = createServiceClient(); + + const { data: existing } = await svc + .from("messages") + .select("id, conversations!inner(channel_account_id)") + .eq("provider_message_id", msg.id) + .eq("conversations.channel_account_id", accountId) + .maybeSingle(); + if (existing) return null; + + const subject = msg.subject ?? ""; + const fromEmail = msg.from?.emailAddress?.address?.toLowerCase() ?? ""; + const fromName = msg.from?.emailAddress?.name ?? undefined; + const toAddrs = addressListOf(msg.toRecipients); + const ccAddrs = addressListOf(msg.ccRecipients); + const bccAddrs = addressListOf(msg.bccRecipients); + const direction: "inbound" | "outbound" = + fromEmail === acct.address.toLowerCase() ? "outbound" : "inbound"; + const sentAtIso = + msg.receivedDateTime ?? + msg.sentDateTime ?? + new Date().toISOString(); + const snippet = (msg.bodyPreview ?? "").slice(0, 240); + const bodyHtml = + msg.body?.contentType === "html" ? msg.body.content : null; + const bodyText = + msg.body?.contentType === "text" ? msg.body.content : null; + const inReplyTo = getOutlookHeader(msg, "In-Reply-To"); + const references = getOutlookHeader(msg, "References"); + const messageIdHeader = + msg.internetMessageId ?? getOutlookHeader(msg, "Message-ID"); + + const conv = await upsertConversationGeneric({ + accountId, + orgId: acct.org_id, + providerThreadId: msg.conversationId, + subject, + accountAddress: acct.address, + direction, + fromEmail, + fromName, + toAddrs, + ccAddrs, + snippet, + sentAtIso, + isRead: msg.isRead ?? false, + channel: "outlook", + }); + + const { data: inserted, error: msgErr } = await svc + .from("messages") + .insert({ + org_id: acct.org_id, + conversation_id: conv.id, + channel: "outlook", + direction, + from_address: fromEmail, + from_name: fromName ?? null, + to_addresses: toAddrs, + cc_addresses: ccAddrs, + bcc_addresses: bccAddrs, + subject, + body_text: bodyText, + body_html: bodyHtml, + snippet, + provider_message_id: msg.id, + provider_thread_id: msg.conversationId, + in_reply_to: inReplyTo ?? null, + references_chain: references + ? references.split(/\s+/).filter(Boolean) + : [], + sent_at: sentAtIso, + is_read: msg.isRead ?? false, + metadata: { message_id_header: messageIdHeader ?? null }, + }) + .select("id") + .maybeSingle(); + if (msgErr || !inserted) { + throw new Error(`message insert failed: ${msgErr?.message ?? "no row"}`); + } + + let attachmentsStored = 0; + if (msg.hasAttachments) { + try { + const list = await listOutlookAttachments({ + accountId, + messageId: msg.id, + }); + for (const a of list.value ?? []) { + try { + const bytes = await getOutlookAttachmentContent({ + accountId, + messageId: msg.id, + attachmentId: a.id, + }); + const path = `${acct.org_id}/${accountId}/${inserted.id}/${a.name}`; + const { error: upErr } = await svc.storage + .from("email-attachments") + .upload(path, bytes, { + contentType: a.contentType, + upsert: true, + }); + if (upErr && !upErr.message?.includes("already")) { + throw upErr; + } + await svc.from("message_attachments").insert({ + org_id: acct.org_id, + message_id: inserted.id, + filename: a.name, + mime_type: a.contentType, + size_bytes: a.size, + storage_path: path, + provider_attachment_id: a.id, + is_inline: a.isInline, + }); + attachmentsStored += 1; + } catch (e) { + console.error("outlook attachment store failed", e); + } + } + } catch (e) { + console.error("outlook attachment list failed", e); + } + } + + await svc + .from("conversations") + .update({ + last_message_at: sentAtIso, + last_message_preview: snippet, + unread_count: + direction === "inbound" && !(msg.isRead ?? false) + ? (conv.unread_count ?? 0) + 1 + : conv.unread_count ?? 0, + }) + .eq("id", conv.id); + + return { attachments: attachmentsStored }; +} + +async function upsertConversationGeneric(opts: { + accountId: string; + orgId: string; + providerThreadId: string; + subject: string; + accountAddress: string; + direction: "inbound" | "outbound"; + fromEmail: string; + fromName?: string; + toAddrs: string[]; + ccAddrs: string[]; + snippet: string; + sentAtIso: string; + isRead: boolean; + channel: "gmail" | "outlook"; +}): Promise<{ id: string; unread_count: number | null }> { + const svc = createServiceClient(); + const { data: existing } = await svc + .from("conversations") + .select("id, participants, lead_id, unread_count") + .eq("channel_account_id", opts.accountId) + .eq("provider_thread_key", opts.providerThreadId) + .maybeSingle(); + + const counterparty = + opts.direction === "inbound" + ? opts.fromEmail + : opts.toAddrs.find( + (a) => a.toLowerCase() !== opts.accountAddress.toLowerCase() + ) ?? null; + + if (existing) { + const known = new Set( + ((existing.participants as { address: string }[]) ?? []).map( + (p) => p.address + ) + ); + const merged = [ + ...((existing.participants as unknown[]) ?? []), + ] as { address: string; name?: string }[]; + const candidates: { address: string; name?: string }[] = [ + { address: opts.fromEmail, name: opts.fromName }, + ...opts.toAddrs.map((a) => ({ address: a })), + ...opts.ccAddrs.map((a) => ({ address: a })), + ]; + for (const c of candidates) { + if (c.address && !known.has(c.address)) { + merged.push(c); + known.add(c.address); + } + } + await svc + .from("conversations") + .update({ participants: merged }) + .eq("id", existing.id); + return { + id: existing.id as string, + unread_count: existing.unread_count as number | null, + }; + } + + let leadId: string | null = null; + if (counterparty) { + leadId = await findLeadForAddress(svc, { + orgId: opts.orgId, + address: counterparty, + }); + } + + const initialParticipants: { address: string; name?: string }[] = [ + { address: opts.fromEmail, name: opts.fromName }, + ...opts.toAddrs.map((a) => ({ address: a })), + ...opts.ccAddrs.map((a) => ({ address: a })), + ].filter((p) => p.address); + + const { data: inserted, error } = await svc + .from("conversations") + .insert({ + org_id: opts.orgId, + channel_account_id: opts.accountId, + channel: opts.channel, + provider_thread_key: opts.providerThreadId, + subject: opts.subject, + lead_id: leadId, + participants: initialParticipants, + last_message_at: opts.sentAtIso, + last_message_preview: opts.snippet.slice(0, 240), + unread_count: + opts.direction === "inbound" && !opts.isRead ? 1 : 0, + }) + .select("id, unread_count") + .maybeSingle(); + if (error || !inserted) { + throw new Error( + `conversation insert failed: ${error?.message ?? "no row"}` + ); + } + return { + id: inserted.id as string, + unread_count: inserted.unread_count as number | null, + }; +} + +// Convenience used by the cron route — syncs every active gmail and outlook +// account. export async function syncAllActiveAccounts(): Promise { const svc = createServiceClient(); const { data: accounts } = await svc .from("channel_accounts") - .select("id") - .eq("provider", "gmail") + .select("id, provider") + .in("provider", ["gmail", "outlook"]) .eq("status", "active"); const results: SyncResult[] = []; for (const a of accounts ?? []) { try { - results.push(await syncGmailAccount(a.id)); + if (a.provider === "gmail") { + results.push(await syncGmailAccount(a.id as string)); + } else if (a.provider === "outlook") { + results.push(await syncOutlookAccount(a.id as string)); + } } catch (e) { results.push({ accountId: a.id as string,