feat(users): add endpoint to update users' marketing updates by email

app/api/public.py

@@ -739,7 +739,7 @@ def _normalize_bedrock_provider_id(
 
     normalized = provider_model_id.split("/", 1)[1]
     if normalized.startswith(provider_prefix):
-        normalized = normalized[len(provider_prefix):]
+        normalized = normalized[len(provider_prefix) :]
     return normalized
 
 
@@ -778,9 +778,7 @@ async def _collect_region_bedrock_models(
         if not isinstance(item, dict):
             continue
         params = item.get("litellm_params") or {}
-        provider_model_id = (
-            params.get("model") if isinstance(params, dict) else None
-        )
+        provider_model_id = params.get("model") if isinstance(params, dict) else None
         if not isinstance(provider_model_id, str) or not provider_model_id:
             continue
 
@@ -900,8 +898,7 @@ async def list_missing_provider_models(
         raise HTTPException(
             status_code=404,
             detail=(
-                f"Unknown provider '{provider}'. "
-                f"Supported: {sorted(_KNOWN_PROVIDERS)}"
+                f"Unknown provider '{provider}'. Supported: {sorted(_KNOWN_PROVIDERS)}"
             ),
         )
 

app/api/users.py

@@ -29,6 +29,7 @@
     UserSpendRegion,
     UserSpendByEmailResponse,
     UserSpendTeam,
+    UserMarketingUpdatesByEmailUpdate,
 )
 from app.db.models import (
     DBPrivateAIKey,
@@ -48,6 +49,7 @@
 )
 from app.core.roles import UserRole
 from app.services.litellm import LiteLLMService
+from app.services.hubspot import HubSpotService
 from datetime import datetime, UTC
 import logging
 import asyncio
@@ -505,6 +507,48 @@ async def get_users_by_email(
     return result
 
 
+@router.put(
+    "/by-email/marketing-updates",
+    response_model=List[User],
+    dependencies=[Depends(get_role_min_system_admin)],
+)
+async def update_users_marketing_updates_by_email(
+    payload: UserMarketingUpdatesByEmailUpdate,
+    db: Session = Depends(get_db),
+):
+    normalized_email = _normalize_email_for_lookup(payload.email)
+
+    users = (
+        db.query(DBUser)
+        .outerjoin(DBTeam, DBUser.team_id == DBTeam.id)
+        .filter(
+            func.regexp_replace(func.lower(DBUser.email), r"\+[^@]*@", "@")
+            == normalized_email,
+            DBUser.is_active.is_(True),
+            (DBUser.team_id.is_(None)) | (DBTeam.deleted_at.is_(None)),
+        )
+        .all()
+    )
+    if not users:
+        return []
+
+    for user in users:
+        user.receive_marketing_updates = payload.receive_marketing_updates
+    db.commit()
+    for user in users:
+        db.refresh(user)
+
+    hubspot = HubSpotService()
+    try:
+        await hubspot.upsert_contacts_marketable_status(
+            [(normalized_email, user.receive_marketing_updates) for user in users]
+        )
+    except HTTPException:
+        logger.exception("HubSpot sync failed for users marketing-updates by email")
+
+    return users
+
+
 @router.get(
     "/spend",
     response_model=UserSpendByEmailResponse,
@@ -754,6 +798,11 @@ async def _create_user_in_db(user: UserCreate, db: Session) -> DBUser:
         is_admin=False,  # Users are created as non-admin by default
         team_id=user.team_id,
         role=user.role,
+        receive_marketing_updates=(
+            user.receive_marketing_updates
+            if user.receive_marketing_updates is not None
+            else False
+        ),
     )
 
     db.add(db_user)
@@ -836,6 +885,7 @@ async def update_user(
             )
 
     previous_email = db_user.email
+    previous_marketing_updates = db_user.receive_marketing_updates
     for key, value in user_update.model_dump(exclude_unset=True).items():
         setattr(db_user, key, value)
 
@@ -904,6 +954,22 @@ async def update_user(
         raise
 
     db.refresh(db_user)
+
+    if (
+        user_update.receive_marketing_updates is not None
+        and user_update.receive_marketing_updates != previous_marketing_updates
+    ):
+        hubspot = HubSpotService()
+        try:
+            await hubspot.upsert_contact_marketable_status(
+                email=db_user.email, enabled=db_user.receive_marketing_updates
+            )
+        except HTTPException:
+            logger.exception(
+                "HubSpot sync failed for user marketing-updates update user_id=%s",
+                db_user.id,
+            )
+
     return db_user
 
 

app/core/config.py

@@ -47,6 +47,7 @@ class Settings(BaseSettings):
     STRIPE_SECRET_KEY: str = os.getenv("STRIPE_SECRET_KEY", "sk_test_string")
     STRIPE_PUBLISHABLE_KEY: str = os.getenv("STRIPE_PUBLISHABLE_KEY", "pk_test_string")
     WEBHOOK_SIG: str = os.getenv("WEBHOOK_SIG", "whsec_test_1234567890")
+    HUBSPOT_TOKEN: str = os.getenv("HUBSPOT_TOKEN", "")
     ENABLE_METRICS: bool = os.getenv("ENABLE_METRICS", "false") == "true"
     PROMETHEUS_API_KEY: str = os.getenv("PROMETHEUS_API_KEY", "")
     POOL_BUDGET_EXPIRATION_DAYS: int = int(

app/db/models.py

@@ -147,6 +147,9 @@ class DBUser(Base):
     is_admin = Column(Boolean, default=False)
     created_at = Column(DateTime(timezone=True), default=func.now())
     role = Column(String, default="user")  # user, admin, key_creator, read_only
+    receive_marketing_updates = Column(
+        Boolean, default=False, nullable=False, server_default=text("false")
+    )
     team_id = Column(Integer, ForeignKey("teams.id", name="fk_user_team"))
     updated_at = Column(DateTime(timezone=True), onupdate=func.now())
 

app/main.py

@@ -256,17 +256,14 @@ def custom_openapi():
                     del operation["parameters"]
 
             # Remove security from non-protected endpoints
-            if (
-                path_name
-                in [
-                    "/auth/login",
-                    "/auth/register",
-                    "/health",
-                    "/auth/generate-trial-access",
-                    "/public/models",
-                    "/public/models/",
-                ]
-            ):
+            if path_name in [
+                "/auth/login",
+                "/auth/register",
+                "/health",
+                "/auth/generate-trial-access",
+                "/public/models",
+                "/public/models/",
+            ]:
                 if "security" in operation:
                     del operation["security"]
 

app/migrations/versions/20260520_160000_add_receive_marketing_updates_to_users.py

@@ -0,0 +1,32 @@
+"""add receive_marketing_updates to users
+
+Revision ID: c4a9d8e1f2b3
+Revises: 2f7c9d1e4aab
+Create Date: 2026-05-20 16:00:00.000000
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = "c4a9d8e1f2b3"
+down_revision = "2f7c9d1e4aab"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "users",
+        sa.Column(
+            "receive_marketing_updates",
+            sa.Boolean(),
+            nullable=False,
+            server_default=sa.text("false"),
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("users", "receive_marketing_updates")

app/schemas/models.py

@@ -47,12 +47,14 @@ class UserCreate(UserBase):
     password: Optional[str] = None
     team_id: Optional[int] = None
     role: Optional[str] = None
+    receive_marketing_updates: Optional[bool] = None
     model_config = ConfigDict(from_attributes=True)
 
 
 class UserUpdate(BaseModel):
     email: Optional[CaseInsensitiveEmailStr] = None
     is_admin: Optional[bool] = None
+    receive_marketing_updates: Optional[bool] = None
     current_password: Optional[str] = None
     new_password: Optional[str] = None
 
@@ -64,10 +66,16 @@ class User(UserBase):
     team_id: Optional[int] = None
     team_name: Optional[str] = None
     role: Optional[str] = None
+    receive_marketing_updates: bool = False
     model_config = ConfigDict(from_attributes=True)
     audit_logs: ClassVar = relationship("AuditLog", back_populates="user")
 
 
+class UserMarketingUpdatesByEmailUpdate(BaseModel):
+    email: CaseInsensitiveEmailStr
+    receive_marketing_updates: bool
+
+
 class APITokenBase(BaseModel):
     name: str
 

app/services/hubspot.py

@@ -0,0 +1,74 @@
+import logging
+from typing import Optional
+
+import httpx
+from fastapi import HTTPException, status
+
+from app.core.config import settings
+
+logger = logging.getLogger(__name__)
+
+
+class HubSpotService:
+    BASE_URL = "https://api.hubapi.com"
+
+    def __init__(self, token: Optional[str] = None):
+        self.token = token or settings.HUBSPOT_TOKEN
+
+    def _headers(self) -> dict[str, str]:
+        if not self.token:
+            raise HTTPException(
+                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                detail="HubSpot token is not configured",
+            )
+        return {
+            "Authorization": f"Bearer {self.token}",
+            "Content-Type": "application/json",
+        }
+
+    async def _upsert_contacts_marketable_status(
+        self, contacts: list[tuple[str, bool]]
+    ) -> None:
+        if not contacts:
+            return
+        payload = {
+            "inputs": [
+                {
+                    "idProperty": "email",
+                    "id": email,
+                    "properties": {
+                        "email": email,
+                        "hs_marketable_status": "true" if enabled else "false",
+                    },
+                }
+                for email, enabled in contacts
+            ]
+        }
+        async with httpx.AsyncClient(timeout=10.0) as client:
+            response = await client.post(
+                f"{self.BASE_URL}/crm/v3/objects/contacts/batch/upsert",
+                headers=self._headers(),
+                json=payload,
+            )
+
+        if response.status_code >= 400:
+            request_id = response.headers.get("x-hubspot-request-id", "unknown")
+            body_excerpt = response.text[:500]
+            logger.error(
+                "HubSpot contact upsert failed status=%s request_id=%s body_excerpt=%s",
+                response.status_code,
+                request_id,
+                body_excerpt,
+            )
+            raise HTTPException(
+                status_code=status.HTTP_502_BAD_GATEWAY,
+                detail="Failed to upsert HubSpot contact",
+            )
+
+    async def upsert_contact_marketable_status(self, email: str, enabled: bool) -> None:
+        await self._upsert_contacts_marketable_status([(email, enabled)])
+
+    async def upsert_contacts_marketable_status(
+        self, contacts: list[tuple[str, bool]]
+    ) -> None:
+        await self._upsert_contacts_marketable_status(contacts)

scripts/check_missing_models.py

@@ -75,7 +75,9 @@ def save_state(state_file: Path, state: dict[str, Any]) -> None:
     state_file.write_text(json.dumps(state, indent=2) + "\n", encoding="utf-8")
 
 
-def build_diff(report: dict[str, Any], previous_state: dict[str, Any]) -> dict[str, Any]:
+def build_diff(
+    report: dict[str, Any], previous_state: dict[str, Any]
+) -> dict[str, Any]:
     """Compute per-region-group ``new_missing`` vs. ``previous_state`` plus
     the next state to persist.
 
@@ -158,7 +160,9 @@ def write_github_outputs(diff: dict[str, Any]) -> None:
         return
     summary = diff["slack_summary"] or "No new missing models detected."
     with open(output_path, "a", encoding="utf-8") as handle:
-        handle.write(f"has_new_missing={'true' if diff['has_new_missing'] else 'false'}\n")
+        handle.write(
+            f"has_new_missing={'true' if diff['has_new_missing'] else 'false'}\n"
+        )
         handle.write(f"state_changed={'true' if diff['state_changed'] else 'false'}\n")
         handle.write(f"provider={diff.get('provider', '')}\n")
         # JSON-encoded so newlines/quotes are safe to interpolate into a Slack payload.