feat: add Google Analytics 4 (GA4) integration

backend/app/main.py

@@ -22,6 +22,7 @@
 from app.routers import template_router
 from app.routers import hubspot_router
 from app.routers import salesforce_router
+from app.routers import ga4_router
 from app.models import User
 from app.auth import hash_password, GUEST_USER_ID, ADMIN_USER_ID
 
@@ -247,6 +248,7 @@ async def httpx_connect_error_handler(_request: Request, exc: httpx.ConnectError
 app.include_router(template_router.router, prefix=prefix)
 app.include_router(hubspot_router.router, prefix=prefix)
 app.include_router(salesforce_router.router, prefix=prefix)
+app.include_router(ga4_router.router, prefix=prefix)
 app.include_router(pipedrive_router.router, prefix=prefix)
 
 

backend/app/routers/ask.py

@@ -36,6 +36,7 @@
 from app.scripts.ask_snowflake import ask_snowflake
 from app.scripts.ask_stripe import ask_stripe
 from app.scripts.ask_pipedrive import ask_pipedrive
+from app.scripts.ask_ga4 import ask_ga4
 from app.scripts.ask_salesforce import ask_salesforce
 from app.scripts.ask_sql import ask_sql
 from app.scripts.ask_sql_multi import ask_sql_multi_source
@@ -523,6 +524,26 @@ async def dispatch_question(
             history=history,
             channel=channel,
         )
+    elif source.type == "ga4":
+        meta = source.metadata_ or {}
+        ga4_creds = meta.get("credentialsContent", "")
+        ga4_prop = meta.get("propertyId", "")
+        ga4_tables = meta.get("tables") or None
+        ga4_table_infos = meta.get("tableInfos") or None
+        if not ga4_creds or not ga4_prop:
+            raise HTTPException(400, "GA4 source missing credentialsContent or propertyId in metadata")
+        result = await ask_ga4(
+            credentials_content=ga4_creds,
+            property_id=ga4_prop,
+            tables=ga4_tables,
+            question=question,
+            agent_description=agent.description or "",
+            source_name=source.name,
+            table_infos=ga4_table_infos,
+            llm_overrides=llm_overrides,
+            history=history,
+            channel=channel,
+        )
     else:
         raise HTTPException(400, f"Unsupported source type: {source.type}")
 

backend/app/routers/crud.py

@@ -135,7 +135,7 @@ async def create_source(
     user: User = Depends(require_user),
 ):
     """Create a non-file source (BigQuery, Google Sheets, SQL). Credentials stored locally in metadata."""
-    valid_types = ("bigquery", "google_sheets", "sql_database", "firebase", "mongodb", "snowflake", "notion", "excel_online", "s3", "rest_api", "jira", "hubspot", "stripe", "pipedrive", "salesforce")
+    valid_types = ("bigquery", "google_sheets", "sql_database", "firebase", "mongodb", "snowflake", "notion", "excel_online", "s3", "rest_api", "jira", "hubspot", "stripe", "pipedrive", "salesforce", "ga4")
     if body.type not in valid_types:
         raise HTTPException(400, f"type must be one of: {', '.join(valid_types)}")
     source_id = str(uuid.uuid4())

backend/app/routers/ga4_router.py

@@ -0,0 +1,120 @@
+"""
+Google Analytics 4 (GA4) router: test connection, discover tables, refresh metadata.
+"""
+import asyncio
+import json
+from fastapi import APIRouter, Depends, HTTPException
+from app.auth import require_user
+from app.models import User, Source
+from app.database import get_db, AsyncSessionLocal
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.routers.crud import _sanitize_for_json
+
+router = APIRouter(prefix="/ga4", tags=["ga4"])
+
+
+@router.post("/test-connection")
+async def test_connection(
+    body: dict,
+    user: User = Depends(require_user),
+):
+    """Test GA4 connection by running a minimal report.
+    Body: { "credentialsContent": "...", "propertyId": "..." }
+    """
+    credentials_content = body.get("credentialsContent") or body.get("credentials_content")
+    property_id = body.get("propertyId") or body.get("property_id")
+
+    if not credentials_content:
+        raise HTTPException(400, "credentialsContent is required")
+    if not property_id:
+        raise HTTPException(400, "propertyId is required")
+
+    from app.scripts.ask_ga4 import _get_access_token, _test_connection_sync
+
+    try:
+        sa = json.loads(credentials_content)
+    except json.JSONDecodeError:
+        raise HTTPException(400, "Invalid JSON in credentialsContent")
+
+    try:
+        access_token = await _get_access_token(sa)
+    except Exception as e:
+        raise HTTPException(400, f"Authentication failed: {e}")
+
+    loop = asyncio.get_event_loop()
+    try:
+        result = await loop.run_in_executor(
+            None, lambda: _test_connection_sync(access_token, property_id)
+        )
+    except Exception as e:
+        raise HTTPException(400, f"Connection test failed: {e}")
+
+    return result
+
+
+@router.post("/discover")
+async def discover_tables(
+    body: dict,
+    user: User = Depends(require_user),
+):
+    """Discover GA4 tables (fetch sample data for each).
+    Body: { "credentialsContent": "...", "propertyId": "...", "tables": [...] (optional) }
+    """
+    credentials_content = body.get("credentialsContent") or body.get("credentials_content")
+    property_id = body.get("propertyId") or body.get("property_id")
+    tables = body.get("tables")
+
+    if not credentials_content:
+        raise HTTPException(400, "credentialsContent is required")
+    if not property_id:
+        raise HTTPException(400, "propertyId is required")
+
+    from app.scripts.ask_ga4 import discover_ga4
+
+    try:
+        table_infos = await discover_ga4(credentials_content, property_id, tables)
+    except Exception as e:
+        raise HTTPException(400, f"Discovery failed: {e}")
+
+    return {"table_infos": table_infos}
+
+
+@router.post("/sources/{source_id}/refresh-metadata")
+async def refresh_source_metadata(
+    source_id: str,
+    db: AsyncSession = Depends(get_db),
+    user: User = Depends(require_user),
+):
+    """Refresh GA4 source metadata (re-fetch table schemas and preview data)."""
+    r = await db.execute(select(Source).where(Source.id == source_id, Source.user_id == user.id))
+    source = r.scalar_one_or_none()
+    if not source:
+        raise HTTPException(404, "Source not found")
+    if source.type != "ga4":
+        raise HTTPException(400, "Source is not GA4")
+
+    meta = dict(source.metadata_ or {})
+    creds = meta.get("credentialsContent") or meta.get("credentials_content")
+    if not creds:
+        raise HTTPException(400, "Source has no credentials")
+
+    property_id = meta.get("propertyId") or meta.get("property_id")
+    if not property_id:
+        raise HTTPException(400, "Source has no propertyId")
+
+    tables = meta.get("tables")
+
+    from app.scripts.ask_ga4 import discover_ga4
+
+    try:
+        table_infos = await discover_ga4(creds, property_id, tables)
+    except Exception as e:
+        raise HTTPException(400, str(e))
+
+    meta["table_infos"] = table_infos
+    source.metadata_ = _sanitize_for_json(meta)
+    await db.commit()
+    await db.refresh(source)
+    return {"metaJSON": source.metadata_}