Feat: HubSpot + Salesforce CRM integrations (Priority 4 complete)

Backend:
- services/hubspot_service.py: upsert contact + create note via Private App token
- services/salesforce_service.py: OAuth ROPC auth, upsert Contact, create Task
- routes/crm_integrations.py: connect/status/test/disconnect for both CRMs
- ai_processor.py: Step 8 — fire CRM sync after email processing (skip spam/newsletter)
- main.py: register crm_integrations router

Frontend:
- HubSpotIntegration + SalesforceIntegration components in settings/index.tsx
- crmApi client in lib/api.ts (hubspot + salesforce namespaced methods)
- Both cards shown in Settings → Integrations tab with connect/test/disconnect

Migration:
- migrations_v4.sql: hubspot_* + sf_* columns on user_profiles

Note: Daily digest (8am UTC) and deadline reminders (9am UTC) were already
fully built in digest_worker.py and email_listener.py.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: 2 people

backend/main.py

@@ -15,7 +15,7 @@
 from slowapi.util import get_remote_address
 
 from config import settings
-from routes import auth, emails, actions, replies, integrations, settings as settings_routes, billing, contacts, outlook, calendar, teams, webhooks as webhooks_routes, autoassign
+from routes import auth, emails, actions, replies, integrations, settings as settings_routes, billing, contacts, outlook, calendar, teams, webhooks as webhooks_routes, autoassign, crm_integrations
 from workers.email_listener import start_email_listener, stop_email_listener
 
 # ---------------------------------------------------------------------------
@@ -151,6 +151,7 @@ async def generic_exception_handler(request: Request, exc: Exception):
 app.include_router(teams.router, prefix="/api")
 app.include_router(webhooks_routes.router, prefix="/api")
 app.include_router(autoassign.router, prefix="/api")
+app.include_router(crm_integrations.router, prefix="/api")
 
 
 # ---------------------------------------------------------------------------

backend/routes/crm_integrations.py

@@ -0,0 +1,159 @@
+"""
+CRM integration settings — HubSpot and Salesforce connection management.
+"""
+import logging
+from typing import Annotated, Optional
+
+from fastapi import APIRouter, Depends, HTTPException
+from pydantic import BaseModel
+
+from database import get_supabase
+from middleware.auth import get_current_user
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/integrations/crm", tags=["crm-integrations"])
+
+
+# ─── Pydantic models ──────────────────────────────────────────────────────────
+
+class HubSpotSaveBody(BaseModel):
+    api_key: str
+
+
+class SalesforceSaveBody(BaseModel):
+    consumer_key: str
+    consumer_secret: str
+    username: str
+    password: str
+    security_token: str
+
+
+# ─── HubSpot endpoints ────────────────────────────────────────────────────────
+
+@router.get("/hubspot/status")
+async def hubspot_status(current_user: Annotated[dict, Depends(get_current_user)]):
+    uid = current_user["id"]
+    supabase = get_supabase()
+    row = supabase.table("user_profiles").select("hubspot_connected, hubspot_api_key").eq("id", uid).single().execute()
+    data = row.data or {}
+    return {
+        "connected": bool(data.get("hubspot_connected")),
+        "has_key": bool(data.get("hubspot_api_key")),
+    }
+
+
+@router.post("/hubspot/connect")
+async def hubspot_connect(body: HubSpotSaveBody, current_user: Annotated[dict, Depends(get_current_user)]):
+    """Save HubSpot Private App token and verify it works."""
+    from services.hubspot_service import test_connection
+
+    uid = current_user["id"]
+    ok = await test_connection(body.api_key)
+    if not ok:
+        raise HTTPException(status_code=400, detail="Invalid HubSpot API key — connection test failed.")
+
+    supabase = get_supabase()
+    supabase.table("user_profiles").update({
+        "hubspot_api_key": body.api_key,
+        "hubspot_connected": True,
+    }).eq("id", uid).execute()
+    return {"connected": True}
+
+
+@router.post("/hubspot/test")
+async def hubspot_test(current_user: Annotated[dict, Depends(get_current_user)]):
+    """Send a test ping to HubSpot."""
+    from services.hubspot_service import test_connection
+
+    uid = current_user["id"]
+    supabase = get_supabase()
+    row = supabase.table("user_profiles").select("hubspot_api_key").eq("id", uid).single().execute()
+    api_key = (row.data or {}).get("hubspot_api_key", "")
+    if not api_key:
+        raise HTTPException(status_code=400, detail="HubSpot not connected.")
+
+    ok = await test_connection(api_key)
+    return {"success": ok}
+
+
+@router.delete("/hubspot/disconnect")
+async def hubspot_disconnect(current_user: Annotated[dict, Depends(get_current_user)]):
+    uid = current_user["id"]
+    supabase = get_supabase()
+    supabase.table("user_profiles").update({
+        "hubspot_api_key": None,
+        "hubspot_connected": False,
+    }).eq("id", uid).execute()
+    return {"connected": False}
+
+
+# ─── Salesforce endpoints ─────────────────────────────────────────────────────
+
+@router.get("/salesforce/status")
+async def salesforce_status(current_user: Annotated[dict, Depends(get_current_user)]):
+    uid = current_user["id"]
+    supabase = get_supabase()
+    row = supabase.table("user_profiles").select("sf_connected, sf_username").eq("id", uid).single().execute()
+    data = row.data or {}
+    return {
+        "connected": bool(data.get("sf_connected")),
+        "username": data.get("sf_username"),
+    }
+
+
+@router.post("/salesforce/connect")
+async def salesforce_connect(body: SalesforceSaveBody, current_user: Annotated[dict, Depends(get_current_user)]):
+    """Save Salesforce credentials and verify they work."""
+    from services.salesforce_service import test_connection
+
+    uid = current_user["id"]
+    ok = await test_connection(
+        body.consumer_key, body.consumer_secret,
+        body.username, body.password, body.security_token,
+    )
+    if not ok:
+        raise HTTPException(status_code=400, detail="Salesforce authentication failed. Check your credentials.")
+
+    supabase = get_supabase()
+    supabase.table("user_profiles").update({
+        "sf_consumer_key": body.consumer_key,
+        "sf_consumer_secret": body.consumer_secret,
+        "sf_username": body.username,
+        "sf_password": body.password,
+        "sf_security_token": body.security_token,
+        "sf_connected": True,
+    }).eq("id", uid).execute()
+    return {"connected": True}
+
+
+@router.post("/salesforce/test")
+async def salesforce_test(current_user: Annotated[dict, Depends(get_current_user)]):
+    """Test the saved Salesforce credentials."""
+    from services.salesforce_service import test_connection
+
+    uid = current_user["id"]
+    supabase = get_supabase()
+    row = supabase.table("user_profiles").select(
+        "sf_consumer_key, sf_consumer_secret, sf_username, sf_password, sf_security_token"
+    ).eq("id", uid).single().execute()
+    d = row.data or {}
+    if not d.get("sf_username"):
+        raise HTTPException(status_code=400, detail="Salesforce not connected.")
+
+    ok = await test_connection(
+        d.get("sf_consumer_key", ""), d.get("sf_consumer_secret", ""),
+        d.get("sf_username", ""), d.get("sf_password", ""), d.get("sf_security_token", ""),
+    )
+    return {"success": ok}
+
+
+@router.delete("/salesforce/disconnect")
+async def salesforce_disconnect(current_user: Annotated[dict, Depends(get_current_user)]):
+    uid = current_user["id"]
+    supabase = get_supabase()
+    supabase.table("user_profiles").update({
+        "sf_consumer_key": None, "sf_consumer_secret": None,
+        "sf_username": None, "sf_password": None, "sf_security_token": None,
+        "sf_connected": False,
+    }).eq("id", uid).execute()
+    return {"connected": False}

backend/services/ai_processor.py

@@ -230,7 +230,47 @@ async def process_email(email: dict) -> dict | None:
             )
 
     # ------------------------------------------------------------------
-    # Step 8 – Auto-assign rules
+    # Step 8 – CRM sync (HubSpot / Salesforce)
+    # ------------------------------------------------------------------
+    try:
+        crm_profile = supabase.table("user_profiles").select(
+            "hubspot_connected, hubspot_api_key, "
+            "sf_connected, sf_consumer_key, sf_consumer_secret, "
+            "sf_username, sf_password, sf_security_token"
+        ).eq("id", user_id).single().execute()
+        cp = crm_profile.data or {}
+
+        # Skip CRM sync for spam/newsletters
+        email_category = analysis.get("category", "")
+        if email_category not in ("spam", "newsletter"):
+            if cp.get("hubspot_connected") and cp.get("hubspot_api_key"):
+                from services.hubspot_service import sync_email_to_hubspot
+                await sync_email_to_hubspot(
+                    api_key=cp["hubspot_api_key"],
+                    sender_email=sender,
+                    sender_name=email.get("from_name", ""),
+                    subject=subject,
+                    summary=analysis.get("summary", ""),
+                )
+
+            if cp.get("sf_connected") and cp.get("sf_username"):
+                from services.salesforce_service import sync_email_to_salesforce
+                await sync_email_to_salesforce(
+                    consumer_key=cp.get("sf_consumer_key", ""),
+                    consumer_secret=cp.get("sf_consumer_secret", ""),
+                    username=cp.get("sf_username", ""),
+                    password=cp.get("sf_password", ""),
+                    security_token=cp.get("sf_security_token", ""),
+                    sender_email=sender,
+                    sender_name=email.get("from_name", ""),
+                    subject=subject,
+                    summary=analysis.get("summary", ""),
+                )
+    except Exception as exc:
+        logger.warning("CRM sync failed for email_id=%s: %s", email_id, exc)
+
+    # ------------------------------------------------------------------
+    # Step 9 – Auto-assign rules
     # ------------------------------------------------------------------
     try:
         _apply_auto_assign_rules(

backend/services/hubspot_service.py

@@ -0,0 +1,110 @@
+"""
+HubSpot CRM integration — push contacts and email notes via Private App token.
+Docs: https://developers.hubspot.com/docs/api/crm/contacts
+"""
+import logging
+from typing import Optional
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+HS_BASE = "https://api.hubapi.com"
+
+
+def _headers(api_key: str) -> dict:
+    return {
+        "Authorization": f"Bearer {api_key}",
+        "Content-Type": "application/json",
+    }
+
+
+async def test_connection(api_key: str) -> bool:
+    """Verify that the API key is valid by calling the /crm/v3/owners endpoint."""
+    try:
+        async with httpx.AsyncClient(timeout=10.0) as client:
+            resp = await client.get(f"{HS_BASE}/crm/v3/owners", headers=_headers(api_key))
+        return resp.status_code == 200
+    except Exception as exc:
+        logger.warning("HubSpot test_connection failed: %s", exc)
+        return False
+
+
+async def upsert_contact(api_key: str, email: str, first_name: str = "", last_name: str = "") -> Optional[str]:
+    """
+    Create or update a HubSpot contact by email. Returns the contact ID.
+    """
+    try:
+        async with httpx.AsyncClient(timeout=10.0) as client:
+            # Search for existing contact
+            search_resp = await client.post(
+                f"{HS_BASE}/crm/v3/objects/contacts/search",
+                headers=_headers(api_key),
+                json={
+                    "filterGroups": [{"filters": [{"propertyName": "email", "operator": "EQ", "value": email}]}],
+                    "properties": ["email", "firstname", "lastname"],
+                    "limit": 1,
+                },
+            )
+            if search_resp.status_code == 200:
+                results = search_resp.json().get("results", [])
+                if results:
+                    return results[0]["id"]
+
+            # Create new contact
+            props: dict = {"email": email}
+            if first_name:
+                props["firstname"] = first_name
+            if last_name:
+                props["lastname"] = last_name
+
+            create_resp = await client.post(
+                f"{HS_BASE}/crm/v3/objects/contacts",
+                headers=_headers(api_key),
+                json={"properties": props},
+            )
+            if create_resp.status_code in (200, 201):
+                return create_resp.json()["id"]
+    except Exception as exc:
+        logger.warning("HubSpot upsert_contact failed: %s", exc)
+    return None
+
+
+async def create_note(api_key: str, contact_id: str, subject: str, body: str) -> bool:
+    """Create a note in HubSpot and associate it with the given contact."""
+    try:
+        note_body = f"*{subject}*\n\n{body}"[:65000]
+        async with httpx.AsyncClient(timeout=10.0) as client:
+            resp = await client.post(
+                f"{HS_BASE}/crm/v3/objects/notes",
+                headers=_headers(api_key),
+                json={
+                    "properties": {"hs_note_body": note_body},
+                    "associations": [
+                        {
+                            "to": {"id": contact_id},
+                            "types": [{"associationCategory": "HUBSPOT_DEFINED", "associationTypeId": 202}],
+                        }
+                    ],
+                },
+            )
+        return resp.status_code in (200, 201)
+    except Exception as exc:
+        logger.warning("HubSpot create_note failed: %s", exc)
+        return False
+
+
+async def sync_email_to_hubspot(api_key: str, sender_email: str, sender_name: str, subject: str, summary: str) -> None:
+    """
+    Full sync: upsert contact then attach a note with the email summary.
+    """
+    parts = sender_name.strip().split(" ", 1)
+    first = parts[0] if parts else ""
+    last = parts[1] if len(parts) > 1 else ""
+
+    contact_id = await upsert_contact(api_key, sender_email, first, last)
+    if contact_id:
+        await create_note(api_key, contact_id, subject, summary or "Email received via InboxIQ.")
+        logger.info("HubSpot: synced email from %s (contact=%s)", sender_email, contact_id)
+    else:
+        logger.warning("HubSpot: could not upsert contact for %s", sender_email)