Migrate DuckDB to ClickHouse

.env.example

@@ -9,3 +9,25 @@ AWS_REGION=us-east-1
 AWS_ACCESS_KEY_ID=YOUR_AWS_ACCESS_KEY
 AWS_SECRET_ACCESS_KEY=YOUR_AWS_SECRET_KEY
 AWS_S3_BUCKET=your-gridify-bucket
+
+# --------------------------------------------------------------------------- #
+# Scalability / real-time / AI-safety improvements
+# --------------------------------------------------------------------------- #
+
+# OLAP storage tier (Improvement 1). duckdb | clickhouse | motherduck
+OLAP_BACKEND=duckdb
+# ClickHouse (serverless/distributed OLAP)
+CLICKHOUSE_URL=http://localhost:8123
+CLICKHOUSE_USER=default
+CLICKHOUSE_PASSWORD=
+CLICKHOUSE_DATABASE=gridify
+# MotherDuck (shared cloud DuckDB store)
+MOTHERDUCK_TOKEN=
+MOTHERDUCK_DATABASE=gridify
+
+# Real-time collaborative layout sync (Improvement 2).
+# broadcast | yjs | supabase
+REALTIME_PROVIDER=broadcast
+YJS_WEBSOCKET_URL=
+SUPABASE_URL=
+SUPABASE_ANON_KEY=

README.md

@@ -301,6 +301,63 @@ Recent hardening across the AI, data, and infrastructure layers:
 * **ElastiCache Valkey Cluster:** Set `REDIS_CLUSTER_MODE=true` (with a `rediss://` `REDIS_URL`) to use the cluster client.
 * **Cloud Secrets Management:** `SECRETS_BACKEND=vault|aws` hydrates the environment from HashiCorp Vault or AWS Secrets Manager before settings load (`backend/app/utils/secrets.py`) — zero secrets are stored in the repo.
 
+## 🧩 Recent Architectural Improvements
+
+Four targeted upgrades that remove single-node, single-session, and
+token/initialization bottlenecks:
+
+### 1. Scalable OLAP Storage Tier (DuckDB → ClickHouse / MotherDuck)
+A single file-attached `gridify.duckdb` per pod cannot scale horizontally. The
+analytics tier now sits behind a `BaseOLAPEngine` (`backend/app/services/olap/`)
+with three interchangeable implementations:
+
+* `DuckDBOLAPEngine` — local in-process (default, zero infra).
+* `ClickHouseOLAPEngine` — serverless/distributed OLAP over HTTP (no native driver).
+* `MotherDuckOLAPEngine` — shared cloud DuckDB store (DuckDB wire-compatible).
+
+Select with `OLAP_BACKEND` (`duckdb|clickhouse|motherduck`); the factory
+(`get_olap_engine`) returns a singleton. Routers/agents depend only on the
+interface, so all pods share one synchronized store. Inspect it at
+`GET /api/olap/engine` and run read-only analytics via `GET /api/olap/query`.
+
+### 2. Real-Time Collaborative Layout Sync
+GenAI drafts and Framer-Motion drags used to live only in ephemeral Zustand
+state. `src/lib/realtimeSync.ts` introduces a transport-agnostic
+`RealtimeProvider` plus a `DashboardSync` binder:
+
+* `BroadcastChannelProvider` — zero-dep cross-tab sync (default; in-process hub
+  fallback for SSR/tests).
+* `YjsProvider` — multiplayer CRDT over WebSocket, lazy-loaded so it never
+  bloats the bundle or test env unless selected.
+* `SupabaseProvider` — hosted realtime (Postgres changes) alternative.
+
+Layouts are now durable and shareable via `LayoutRepository`
+(`backend/app/services/layout_repository.py`) — in-memory by default, PostgreSQL
+when `DATABASE_URL` is reachable — exposed at `GET/POST/DELETE /api/layouts`.
+
+### 3. Formalized Semantic Layer (RBAC + Dialect-Aware + Compact Prompts)
+`backend/app/services/semantic_model.py` formalizes the abstraction fed to Gemini
+so the LLM gets a structured description instead of the raw schema:
+
+* **Role-based data security** — `authorize()` rejects any measure/dimension/
+  filter a role may not access *before* SQL is compiled (`analyst` vs `viewer`).
+* **Dialect-aware compile** — one validated logical query emits DuckDB or
+  ClickHouse SQL (`compile(..., dialect=...)`).
+* **Token-efficient context** — `build_prompt_context(role)` emits a compact
+  description, shrinking prompts and constraining the model to its entitlements.
+  Try `GET /api/semantic/context?role=viewer` and `POST /api/semantic/compile`.
+
+### 4. Edge-RAG Tiered Quantization & Streaming
+Downloading the ONNX model + full vector index hurts "Time to First Query" on
+slow networks. Two additions fix this:
+
+* `src/lib/quantization.ts` — scalar `float32 → int8` quantization (and int8
+  cosine) shrinks embedding payloads ~4x with negligible pre-filter loss.
+* `src/lib/modelStreamCache.ts` — chunked Cache Storage API delivery with
+  **resume** (only missing chunks are re-fetched) and an in-memory fallback.
+  The ONNX worker returns quantized embeddings, and `ragClient.ts` caches/restores
+  a quantized index so refreshes skip re-embedding.
+
 ## 🧪 Testing
 
 ```bash

backend/app/config.py

@@ -33,8 +33,30 @@ class Settings(BaseSettings):
     # Database
     DATABASE_URL: str = "postgresql://gridify:gridify_password@localhost:5432/gridify"
 
-    # DuckDB
+    # DuckDB (local in-process engine — used by default and for MotherDuck-style
+    # local-first query paths).
     DUCKDB_PATH: str = "./data/gridify.duckdb"
+
+    # Pluggable OLAP storage tier.
+    # The heavy cloud analytical tier can now be backed by a serverless /
+    # distributed engine instead of a single file-attached DuckDB. ``OLAP_BACKEND``
+    # selects the engine:
+    #   - "duckdb"    : local in-process DuckDB (default, zero infra)
+    #   - "clickhouse": serverless/distributed ClickHouse over HTTP
+    #   - "motherduck": MotherDuck shared cloud store (DuckDB wire-compatible)
+    # Selecting clickhouse / motherduck lets many FastAPI pods share one
+    # synchronized analytical store instead of split local .duckdb files.
+    OLAP_BACKEND: str = "duckdb"
+
+    # ClickHouse (serverless/distributed OLAP).
+    CLICKHOUSE_URL: str = "http://localhost:8123"
+    CLICKHOUSE_USER: str = "default"
+    CLICKHOUSE_PASSWORD: str = ""
+    CLICKHOUSE_DATABASE: str = "gridify"
+
+    # MotherDuck (shared cloud DuckDB store; duckdb connects via the ``md:`` URL).
+    MOTHERDUCK_TOKEN: Optional[str] = None
+    MOTHERDUCK_DATABASE: str = "gridify"
 
     # Vector DB
     CHROMA_HOST: str = "localhost"
@@ -53,6 +75,15 @@ class Settings(BaseSettings):
     LLM_GATEWAY_LANGFUSE_SECRET_KEY: Optional[str] = None
     LLM_GATEWAY_LANGFUSE_HOST: str = "https://cloud.langfuse.com"
 
+    # Real-time collaborative layout sync (Improvement 2).
+    # Selects the transport used by the frontend realtime engine: "broadcast"
+    # (cross-tab, zero-dep), "yjs" (multiplayer CRDT over WebSocket), or
+    # "supabase" (hosted realtime). The frontend provider is chosen to match.
+    REALTIME_PROVIDER: str = "broadcast"
+    YJS_WEBSOCKET_URL: Optional[str] = None
+    SUPABASE_URL: Optional[str] = None
+    SUPABASE_ANON_KEY: Optional[str] = None
+
     # LLM Configuration
     LITELLM_API_KEY: Optional[str] = None
     LLM_PROVIDER: str = "gemini"

backend/app/routers/layouts.py

@@ -0,0 +1,62 @@
+"""Endpoints for saving, loading, and sharing dashboard layouts."""
+
+from __future__ import annotations
+
+import uuid
+from typing import Any, Dict, List
+
+from fastapi import APIRouter, HTTPException, Query
+
+from app.services.layout_repository import (
+    DashboardLayout,
+    get_layout_repository,
+)
+from app.utils.logger import setup_logger
+
+logger = setup_logger(__name__)
+router = APIRouter()
+
+
+@router.post("/layouts")
+async def create_layout(payload: Dict[str, Any]) -> Dict[str, Any]:
+    """Save a new dashboard layout (or overwrite one by id)."""
+    repo = get_layout_repository()
+    layout_id = payload.get("id") or str(uuid.uuid4())
+    layout = DashboardLayout(
+        id=layout_id,
+        name=payload.get("name", "Untitled layout"),
+        owner=payload.get("owner", "anonymous"),
+        widgets=payload.get("widgets", []),
+        is_public=bool(payload.get("is_public", False)),
+    )
+    saved = repo.save(layout)
+    return saved.to_dict()
+
+
+@router.get("/layouts/{layout_id}")
+async def get_layout(layout_id: str) -> Dict[str, Any]:
+    repo = get_layout_repository()
+    layout = repo.get(layout_id)
+    if not layout:
+        raise HTTPException(status_code=404, detail="Layout not found")
+    return layout.to_dict()
+
+
+@router.get("/layouts")
+async def list_layouts(
+    owner: str = Query("anonymous"), public: bool = Query(False)
+) -> Dict[str, Any]:
+    repo = get_layout_repository()
+    layouts: List[DashboardLayout] = (
+        repo.list_public() if public else repo.list_by_owner(owner)
+    )
+    return {"layouts": [l.to_dict() for l in layouts], "count": len(layouts)}
+
+
+@router.delete("/layouts/{layout_id}")
+async def delete_layout(layout_id: str) -> Dict[str, Any]:
+    repo = get_layout_repository()
+    ok = repo.delete(layout_id)
+    if not ok:
+        raise HTTPException(status_code=404, detail="Layout not found")
+    return {"deleted": layout_id}

backend/app/routers/olap.py

@@ -0,0 +1,46 @@
+"""OLAP analytics endpoints backed by the pluggable storage tier."""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, HTTPException, Query
+from typing import Any, Dict, List
+
+from app.services.olap.factory import get_olap_engine
+from app.utils.logger import setup_logger
+
+logger = setup_logger(__name__)
+router = APIRouter()
+
+
+@router.get("/olap/engine")
+async def olap_engine() -> Dict[str, Any]:
+    """Report which OLAP backend is currently active."""
+    engine = get_olap_engine()
+    return {"backend": engine.backend, "healthy": engine.health()}
+
+
+@router.get("/olap/query")
+async def olap_query(
+    sql: str = Query(..., description="Read-only SQL against the analytics tier"),
+    limit: int = Query(100, ge=1, le=1000),
+) -> Dict[str, Any]:
+    """Run a read-only analytics query against the active OLAP backend.
+
+    This is the single entry point used by the GenAI agent and the dashboard so
+    that, regardless of whether the tier is DuckDB, ClickHouse, or MotherDuck,
+    the rest of the app only depends on one interface.
+    """
+    engine = get_olap_engine()
+    try:
+        safe_sql = sql.strip().rstrip(";")
+        if not safe_sql.lower().startswith(("select", "show", "with")):
+            raise HTTPException(
+                status_code=400, detail="Only read-only statements are permitted"
+            )
+        rows = engine.query(f"{safe_sql} LIMIT {limit}")
+        return {"backend": engine.backend, "rows": rows, "count": len(rows)}
+    except HTTPException:
+        raise
+    except Exception as exc:  # pragma: no cover - backend-specific failures
+        logger.warning(f"OLAP query failed: {exc}")
+        raise HTTPException(status_code=502, detail=f"OLAP query failed: {exc}")

backend/app/routers/semantic.py

@@ -0,0 +1,57 @@
+"""Semantic layer endpoints: token-efficient, RBAC-safe prompt context."""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, HTTPException, Query
+from typing import Dict, Any
+
+from app.services.semantic_model import Dialect, get_semantic_model
+from app.utils.logger import setup_logger
+
+logger = setup_logger(__name__)
+router = APIRouter()
+
+
+@router.get("/semantic/context")
+async def semantic_context(
+    role: str = Query("analyst", description="RBAC role requesting context"),
+    dialect: Dialect = Query(Dialect.DUCKDB, description="Target OLAP dialect"),
+) -> Dict[str, Any]:
+    """Return the compact semantic context for a role.
+
+    This structured description (not the raw schema) is what gets injected into
+    the Gemini prompt, shrinking token usage and constraining the model to the
+    role's entitlements.
+    """
+    model = get_semantic_model()
+    try:
+        context = model.build_prompt_context(role, dialect)
+    except Exception as exc:
+        raise HTTPException(status_code=404, detail=str(exc))
+    return {"role": role, "dialect": dialect.value, "context": context}
+
+
+@router.post("/semantic/compile")
+async def semantic_compile(payload: Dict[str, Any]) -> Dict[str, Any]:
+    """Compile an authorized semantic query to engine-specific SQL.
+
+    Body: ``{role, measures, dimensions, filters?, order_by?, limit?, dialect?}``.
+    Raises 403-style 400 when the role is not entitled to part of the request.
+    """
+    model = get_semantic_model()
+    try:
+        sql = model.compile(
+            role=payload["role"],
+            measures=payload.get("measures", []),
+            dimensions=payload.get("dimensions", []),
+            filters=payload.get("filters", []),
+            order_by=payload.get("order_by"),
+            order_direction=payload.get("order_direction", "DESC"),
+            limit=payload.get("limit"),
+            dialect=Dialect(payload.get("dialect", "duckdb")),
+        )
+    except KeyError:
+        raise HTTPException(status_code=400, detail="Missing 'role' in body")
+    except Exception as exc:
+        raise HTTPException(status_code=400, detail=str(exc))
+    return {"sql": sql}