From 9e850d13ad230ccf2dfb44a9c6584754e89e3a33 Mon Sep 17 00:00:00 2001 From: Raymond Oyondi Date: Sun, 12 Jul 2026 18:59:55 +0000 Subject: [PATCH 1/6] Add pluggable OLAP storage tier (DuckDB/ClickHouse/MotherDuck) Resolves the single-node limit: heavy analytical requests across many FastAPI pods now share one synchronized OLAP backend instead of forking into split local gridify.duckdb files. - BaseOLAPEngine interface so routers/agents only depend on one contract - DuckDBOLAPEngine (local-first), ClickHouseOLAPEngine (HTTP), MotherDuckOLAPEngine (shared cloud store) - OLAP_BACKEND config + factory singleton - /api/olap/engine and read-only /api/olap/query endpoints - Unit tests covering binding, literals, and backend selection --- backend/app/config.py | 24 ++- backend/app/routers/olap.py | 46 ++++++ backend/app/services/olap/__init__.py | 33 ++++ backend/app/services/olap/base.py | 41 +++++ .../app/services/olap/clickhouse_engine.py | 142 ++++++++++++++++++ backend/app/services/olap/duckdb_engine.py | 78 ++++++++++ backend/app/services/olap/factory.py | 57 +++++++ .../app/services/olap/motherduck_engine.py | 88 +++++++++++ backend/main.py | 3 +- backend/tests/test_olap.py | 67 +++++++++ 10 files changed, 577 insertions(+), 2 deletions(-) create mode 100644 backend/app/routers/olap.py create mode 100644 backend/app/services/olap/__init__.py create mode 100644 backend/app/services/olap/base.py create mode 100644 backend/app/services/olap/clickhouse_engine.py create mode 100644 backend/app/services/olap/duckdb_engine.py create mode 100644 backend/app/services/olap/factory.py create mode 100644 backend/app/services/olap/motherduck_engine.py create mode 100644 backend/tests/test_olap.py diff --git a/backend/app/config.py b/backend/app/config.py index ec15854..882d600 100644 --- a/backend/app/config.py +++ b/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" diff --git a/backend/app/routers/olap.py b/backend/app/routers/olap.py new file mode 100644 index 0000000..82e50b1 --- /dev/null +++ b/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}") diff --git a/backend/app/services/olap/__init__.py b/backend/app/services/olap/__init__.py new file mode 100644 index 0000000..df52cf6 --- /dev/null +++ b/backend/app/services/olap/__init__.py @@ -0,0 +1,33 @@ +"""Pluggable OLAP storage tier for the Gridify analytics backend. + +The original backend ran DuckDB as an in-process engine. When many FastAPI pods +run heavy analytical AI requests, a single file-attached ``gridify.duckdb`` per +pod leads to split, un-synchronized, duplicated analytical state. + +This module introduces a small engine abstraction so the *heavier* cloud +analytical tier can be backed by a serverless / distributed OLAP engine while +DuckDB remains the local-first engine: + +- ``DuckDBOLAPEngine`` : local in-process DuckDB (default, zero infra) +- ``ClickHouseOLAPEngine`` : serverless/distributed ClickHouse over HTTP +- ``MotherDuckOLAPEngine`` : shared cloud DuckDB store (DuckDB wire-compatible) + +Select the backend with ``OLAP_BACKEND`` (see :mod:`app.config`). +""" + +from __future__ import annotations + +from app.services.olap.base import BaseOLAPEngine +from app.services.olap.clickhouse_engine import ClickHouseOLAPEngine +from app.services.olap.duckdb_engine import DuckDBOLAPEngine +from app.services.olap.factory import OLAPBackend, get_olap_engine +from app.services.olap.motherduck_engine import MotherDuckOLAPEngine + +__all__ = [ + "BaseOLAPEngine", + "ClickHouseOLAPEngine", + "DuckDBOLAPEngine", + "MotherDuckOLAPEngine", + "OLAPBackend", + "get_olap_engine", +] diff --git a/backend/app/services/olap/base.py b/backend/app/services/olap/base.py new file mode 100644 index 0000000..70ed0fe --- /dev/null +++ b/backend/app/services/olap/base.py @@ -0,0 +1,41 @@ +"""Base class for the pluggable OLAP storage engines.""" + +from __future__ import annotations + +from typing import Any, Dict, List, Optional + + +class BaseOLAPEngine: + """Common interface implemented by every OLAP backend. + + The rest of the backend (routers, agents, semantic layer) only depends on + this interface, so swapping the analytical engine never requires touching + call sites. + """ + + backend: str = "base" + + def query(self, sql: str, params: Optional[List[Any]] = None) -> List[Dict[str, Any]]: + """Run SQL and return rows as a list of dicts.""" + raise NotImplementedError + + def query_to_arrow(self, sql: str, params: Optional[List[Any]] = None): + """Run SQL and return an Apache Arrow table. + + Falls back to a row-based conversion when the engine cannot produce a + native Arrow table (e.g. ClickHouse over HTTP). + """ + raise NotImplementedError + + def health(self) -> bool: + """Return ``True`` when the engine can serve queries.""" + raise NotImplementedError + + def close(self) -> None: + """Release any held connections / resources.""" + + def __enter__(self) -> "BaseOLAPEngine": + return self + + def __exit__(self, *exc: Any) -> None: + self.close() diff --git a/backend/app/services/olap/clickhouse_engine.py b/backend/app/services/olap/clickhouse_engine.py new file mode 100644 index 0000000..783c074 --- /dev/null +++ b/backend/app/services/olap/clickhouse_engine.py @@ -0,0 +1,142 @@ +"""Serverless / distributed ClickHouse OLAP engine (HTTP interface). + +ClickHouse gives the FastAPI cluster an industry-standard concurrent multi-user +analytical tier. Every pod talks to the *same* ClickHouse service over HTTP, so +heavy analytical AI requests stay synchronized instead of forking into separate +local ``gridify.duckdb`` files. + +Only the lightweight ClickHouse HTTP interface is used (no native driver), so +the engine is installable with the existing ``httpx`` dependency. +""" + +from __future__ import annotations + +from typing import Any, Dict, List, Optional + +import httpx + +from app.config import settings +from app.services.olap.base import BaseOLAPEngine + + +class ClickHouseOLAPEngine(BaseOLAPEngine): + backend = "clickhouse" + + def __init__( + self, + url: Optional[str] = None, + user: Optional[str] = None, + password: Optional[str] = None, + database: Optional[str] = None, + timeout: float = 30.0, + ): + self.url = (url or settings.CLICKHOUSE_URL).rstrip("/") + self.user = user or settings.CLICKHOUSE_USER + self.password = password or settings.CLICKHOUSE_PASSWORD + self.database = database or settings.CLICKHOUSE_DATABASE + self.timeout = timeout + self._client = httpx.Client( + base_url=self.url, + timeout=self.timeout, + params={"user": self.user, "password": self.password, "database": self.database}, + ) + self._initialize_schema() + + def _query_http(self, sql: str) -> List[Dict[str, Any]]: + resp = self._client.post( + "/", + params={"query": sql, "default_format": "JSONEachRow"}, + ) + resp.raise_for_status() + if not resp.text.strip(): + return [] + return [self._coerce(row) for row in resp.text.strip().split("\n") if row] + + @staticmethod + def _coerce(line: str) -> Dict[str, Any]: + import json + + return json.loads(line) + + def _initialize_schema(self) -> None: + for stmt in ( + """ + CREATE TABLE IF NOT EXISTS telemetry ( + id UInt32, + device_id String, + metric_type String, + value Float64, + timestamp DateTime, + metadata String + ) ENGINE = MergeTree() ORDER BY (device_id, timestamp) + """, + """ + CREATE TABLE IF NOT EXISTS events ( + id UInt32, + event_type String, + source String, + details String, + created_at DateTime + ) ENGINE = MergeTree() ORDER BY (event_type, created_at) + """, + ): + self._client.post("/", params={"query": stmt}).raise_for_status() + + def query(self, sql: str, params: Optional[List[Any]] = None) -> List[Dict[str, Any]]: + if params: + sql = self._bind(sql, params) + return self._query_http(sql) + + def query_to_arrow(self, sql: str, params: Optional[List[Any]] = None): + import pyarrow as pa + + rows = self.query(sql, params) + if not rows: + return pa.table({}) + columns: Dict[str, list] = {k: [] for k in rows[0].keys()} + for row in rows: + for k, v in row.items(): + columns[k].append(v) + return pa.table(columns) + + def health(self) -> bool: + try: + resp = self._client.post("/", params={"query": "SELECT 1"}) + resp.raise_for_status() + return True + except Exception: + return False + + @staticmethod + def _bind(sql: str, params: List[Any]) -> str: + """Positional ``?`` binding for simple literals (ClickHouse HTTP). + + Values are escaped and wrapped as SQL literals. This is sufficient for + telemetry-style parameters; complex payloads should use the native + client with parameterized statements. + """ + out: List[str] = [] + idx = 0 + for ch in sql: + if ch == "?" and idx < len(params): + out.append(ClickHouseOLAPEngine._literal(params[idx])) + idx += 1 + else: + out.append(ch) + return "".join(out) + + @staticmethod + def _literal(value: Any) -> str: + if value is None: + return "NULL" + if isinstance(value, bool): + return "1" if value else "0" + if isinstance(value, (int, float)): + return str(value) + return "'" + str(value).replace("'", "\\'") + "'" + + def close(self) -> None: + try: + self._client.close() + except Exception: + pass diff --git a/backend/app/services/olap/duckdb_engine.py b/backend/app/services/olap/duckdb_engine.py new file mode 100644 index 0000000..c0c782f --- /dev/null +++ b/backend/app/services/olap/duckdb_engine.py @@ -0,0 +1,78 @@ +"""Local DuckDB OLAP engine (default, zero-infra analytical tier).""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any, Dict, List, Optional + +import duckdb + +from app.config import settings +from app.services.olap.base import BaseOLAPEngine + + +class DuckDBOLAPEngine(BaseOLAPEngine): + """In-process DuckDB engine. + + Reuses the existing DuckDB schema bootstrap so local-first behaviour is + unchanged. This is the engine selected by ``OLAP_BACKEND=duckdb``. + """ + + backend = "duckdb" + + def __init__(self, db_path: Optional[str] = None): + db_path = db_path or settings.DUCKDB_PATH + Path(db_path).parent.mkdir(parents=True, exist_ok=True) + self.conn = duckdb.connect(db_path) + self._initialize_schema() + + def _initialize_schema(self) -> None: + self.conn.execute( + """ + CREATE TABLE IF NOT EXISTS telemetry ( + id INTEGER PRIMARY KEY, + device_id VARCHAR, + metric_type VARCHAR, + value DOUBLE, + timestamp TIMESTAMP, + metadata JSON + ) + """ + ) + self.conn.execute( + """ + CREATE TABLE IF NOT EXISTS events ( + id INTEGER PRIMARY KEY, + event_type VARCHAR, + source VARCHAR, + details JSON, + created_at TIMESTAMP + ) + """ + ) + + def query(self, sql: str, params: Optional[List[Any]] = None) -> List[Dict[str, Any]]: + if params: + rows = self.conn.execute(sql, params).fetchall() + else: + rows = self.conn.execute(sql).fetchall() + columns = [d[0] for d in self.conn.description] + return [dict(zip(columns, row)) for row in rows] + + def query_to_arrow(self, sql: str, params: Optional[List[Any]] = None): + if params: + return self.conn.execute(sql, params).fetch_arrow_table() + return self.conn.execute(sql).fetch_arrow_table() + + def health(self) -> bool: + try: + self.conn.execute("SELECT 1").fetchone() + return True + except Exception: + return False + + def close(self) -> None: + try: + self.conn.close() + except Exception: + pass diff --git a/backend/app/services/olap/factory.py b/backend/app/services/olap/factory.py new file mode 100644 index 0000000..2451ce0 --- /dev/null +++ b/backend/app/services/olap/factory.py @@ -0,0 +1,57 @@ +"""Factory selecting the active OLAP backend from configuration.""" + +from __future__ import annotations + +import enum +from typing import Optional + +from app.config import settings +from app.services.olap.base import BaseOLAPEngine + + +class OLAPBackend(str, enum.Enum): + DUCKDB = "duckdb" + CLICKHOUSE = "clickhouse" + MOTHERDUCK = "motherduck" + + +_engines: dict[OLAPBackend, type[BaseOLAPEngine]] = {} + +_engine_instance: Optional[BaseOLAPEngine] = None + + +def _build(backend: OLAPBackend) -> BaseOLAPEngine: + if backend is OLAPBackend.DUCKDB: + from app.services.olap.duckdb_engine import DuckDBOLAPEngine + + return DuckDBOLAPEngine() + if backend is OLAPBackend.CLICKHOUSE: + from app.services.olap.clickhouse_engine import ClickHouseOLAPEngine + + return ClickHouseOLAPEngine() + if backend is OLAPBackend.MOTHERDUCK: + from app.services.olap.motherduck_engine import MotherDuckOLAPEngine + + return MotherDuckOLAPEngine() + raise ValueError(f"Unknown OLAP backend: {backend}") + + +def get_olap_engine(backend: Optional[str] = None) -> BaseOLAPEngine: + """Return the configured OLAP engine singleton. + + Args: + backend: Optional override for ``OLAP_BACKEND`` (used in tests). + """ + global _engine_instance + selected = OLAPBackend((backend or settings.OLAP_BACKEND).lower()) + if _engine_instance is None or _engine_instance.backend != selected.value: + _engine_instance = _build(selected) + return _engine_instance + + +def reset_olap_engine() -> None: + """Drop the cached singleton (used in tests).""" + global _engine_instance + if _engine_instance is not None: + _engine_instance.close() + _engine_instance = None diff --git a/backend/app/services/olap/motherduck_engine.py b/backend/app/services/olap/motherduck_engine.py new file mode 100644 index 0000000..8f8f235 --- /dev/null +++ b/backend/app/services/olap/motherduck_engine.py @@ -0,0 +1,88 @@ +"""MotherDuck shared cloud OLAP engine (DuckDB wire-compatible). + +MotherDuck lets every pod combine local DuckDB querying with a shared, scalable +cloud data store. Because it is DuckDB wire-compatible we reuse the exact same +DuckDB engine — only the connection URL changes to ``md:`` and an +auth token is supplied via ``motherduck_token``. +""" + +from __future__ import annotations + +from typing import Any, Dict, List, Optional + +import duckdb + +from app.config import settings +from app.services.olap.base import BaseOLAPEngine + + +class MotherDuckOLAPEngine(BaseOLAPEngine): + backend = "motherduck" + + def __init__( + self, + database: Optional[str] = None, + token: Optional[str] = None, + ): + database = database or settings.MOTHERDUCK_DATABASE + token = token or settings.MOTHERDUCK_TOKEN + if not token: + raise RuntimeError( + "MOTHERDUCK_TOKEN is required for the MotherDuck OLAP backend" + ) + # DuckDB attaches to MotherDuck over the wire using the ``md:`` URL and + # an auth token kept in a local config file (never logged). + self.conn = duckdb.connect(f"md:{database}") + self.conn.execute(f"CREATE SECRET IF NOT EXISTS md_token (TYPE md, TOKEN '{token}')") + self._initialize_schema() + + def _initialize_schema(self) -> None: + self.conn.execute( + """ + CREATE TABLE IF NOT EXISTS telemetry ( + id INTEGER, + device_id VARCHAR, + metric_type VARCHAR, + value DOUBLE, + timestamp TIMESTAMP, + metadata JSON + ) + """ + ) + self.conn.execute( + """ + CREATE TABLE IF NOT EXISTS events ( + id INTEGER, + event_type VARCHAR, + source VARCHAR, + details JSON, + created_at TIMESTAMP + ) + """ + ) + + def query(self, sql: str, params: Optional[List[Any]] = None) -> List[Dict[str, Any]]: + if params: + rows = self.conn.execute(sql, params).fetchall() + else: + rows = self.conn.execute(sql).fetchall() + columns = [d[0] for d in self.conn.description] + return [dict(zip(columns, row)) for row in rows] + + def query_to_arrow(self, sql: str, params: Optional[List[Any]] = None): + if params: + return self.conn.execute(sql, params).fetch_arrow_table() + return self.conn.execute(sql).fetch_arrow_table() + + def health(self) -> bool: + try: + self.conn.execute("SELECT 1").fetchone() + return True + except Exception: + return False + + def close(self) -> None: + try: + self.conn.close() + except Exception: + pass diff --git a/backend/main.py b/backend/main.py index 161b6ac..3c3ce9d 100644 --- a/backend/main.py +++ b/backend/main.py @@ -16,7 +16,7 @@ from fastapi.staticfiles import StaticFiles from app.config import settings -from app.routers import telemetry, gemini +from app.routers import telemetry, gemini, olap from app.middleware.guardrail_middleware import AsyncGuardrailMiddleware from app.utils.logger import setup_logger @@ -54,6 +54,7 @@ async def lifespan(app: FastAPI): # Include routers app.include_router(telemetry.router, prefix="/api", tags=["telemetry"]) app.include_router(gemini.router, prefix="/api", tags=["gemini"]) +app.include_router(olap.router, prefix="/api", tags=["olap"]) # Async guardrails at the network boundary: prompt-injection scans run off the # synchronous request path (edge service or worker thread) so TTFT stays snappy. diff --git a/backend/tests/test_olap.py b/backend/tests/test_olap.py new file mode 100644 index 0000000..1ca3268 --- /dev/null +++ b/backend/tests/test_olap.py @@ -0,0 +1,67 @@ +"""Tests for the pluggable OLAP storage tier.""" + +import pytest + +from app.services.olap.base import BaseOLAPEngine +from app.services.olap.clickhouse_engine import ClickHouseOLAPEngine +from app.services.olap.duckdb_engine import DuckDBOLAPEngine +from app.services.olap.factory import OLAPBackend, get_olap_engine, reset_olap_engine + + +def test_duckdb_engine_roundtrip(tmp_path): + engine = DuckDBOLAPEngine(db_path=str(tmp_path / "gridify.duckdb")) + engine.query( + "INSERT INTO telemetry (id, device_id, metric_type, value) " + "VALUES (1, 'dev-1', 'temperature', 21.5)" + ) + rows = engine.query("SELECT device_id, value FROM telemetry WHERE id = 1") + assert rows == [{"device_id": "dev-1", "value": 21.5}] + assert engine.health() is True + table = engine.query_to_arrow("SELECT value FROM telemetry") + assert table.num_rows == 1 + engine.close() + + +def test_duckdb_engine_is_base_instance(): + engine = DuckDBOLAPEngine(db_path=":memory:") + assert isinstance(engine, BaseOLAPEngine) + engine.close() + + +def test_clickhouse_engine_binds_parameters(): + assert ( + ClickHouseOLAPEngine._bind("WHERE id = ? AND name = ?", [3, "dev"]) + == "WHERE id = 3 AND name = 'dev'" + ) + + +def test_clickhouse_engine_literal_escaping(): + assert ClickHouseOLAPEngine._literal(None) == "NULL" + assert ClickHouseOLAPEngine._literal(True) == "1" + assert ClickHouseOLAPEngine._literal("a'b") == "'a\\'b'" + + +def test_factory_returns_duckdb_by_default(monkeypatch, tmp_path): + monkeypatch.setattr("app.config.settings.OLAP_BACKEND", "duckdb") + monkeypatch.setattr("app.config.settings.DUCKDB_PATH", str(tmp_path / "g.duckdb")) + reset_olap_engine() + engine = get_olap_engine() + try: + assert engine.backend == OLAPBackend.DUCKDB.value + finally: + reset_olap_engine() + + +def test_factory_selects_clickhouse(monkeypatch): + reset_olap_engine() + engine = get_olap_engine(backend="clickhouse") + assert engine.backend == OLAPBackend.CLICKHOUSE.value + engine.close() + reset_olap_engine() + + +def test_factory_rejects_unknown_backend(): + reset_olap_engine() + with pytest.raises(ValueError): + get_olap_engine(backend="cassandra") + reset_olap_engine() From 270f55122ccef8d49f53fef26db63313b11706eb Mon Sep 17 00:00:00 2001 From: Raymond Oyondi Date: Sun, 12 Jul 2026 19:01:20 +0000 Subject: [PATCH 2/6] Formalize semantic layer with RBAC, dialect compile, and compact prompt context Feeds a structured abstraction (not raw schema) into Gemini to cut token usage and enforce role-based data security. - SemanticModel with logical measures/dimensions and per-role grants - authorize() enforces RBAC before any SQL is compiled - compile() emits engine-specific SQL (DuckDB / ClickHouse dialects) - build_prompt_context() emits a token-efficient description for the LLM - /api/semantic/context and /api/semantic/compile endpoints - Unit tests for RBAC denial, dialect quoting, and context scoping --- backend/app/routers/semantic.py | 57 +++++ backend/app/services/semantic_model.py | 284 +++++++++++++++++++++++++ backend/main.py | 3 +- backend/tests/test_semantic_model.py | 93 ++++++++ 4 files changed, 436 insertions(+), 1 deletion(-) create mode 100644 backend/app/routers/semantic.py create mode 100644 backend/app/services/semantic_model.py create mode 100644 backend/tests/test_semantic_model.py diff --git a/backend/app/routers/semantic.py b/backend/app/routers/semantic.py new file mode 100644 index 0000000..d02e30c --- /dev/null +++ b/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} diff --git a/backend/app/services/semantic_model.py b/backend/app/services/semantic_model.py new file mode 100644 index 0000000..9b06794 --- /dev/null +++ b/backend/app/services/semantic_model.py @@ -0,0 +1,284 @@ +"""Formalized semantic layer (Ibis/Cube-style) with role-based security. + +The heuristic :mod:`app.services.semantic_layer` already stops the LLM from +emitting raw SQL. This module *formalizes* the abstraction so the GenAI +pipeline can: + +1. Feed a structured, token-efficient description of the data model to the LLM + instead of the raw database schema (shorter prompts, fewer tokens). +2. Enforce role-based data security — a ``viewer`` can never request a measure + or dimension they are not entitled to, even if the model attempts it. +3. Compile one validated logical query to engine-specific SQL via a small + dialect layer, so the same semantic query runs on DuckDB, ClickHouse, or + MotherDuck without the model knowing which engine is live. + +The model exposes logical objects (measures/dimensions) and grants per role. +""" + +from __future__ import annotations + +import enum +from dataclasses import dataclass, field +from typing import Any, Dict, FrozenSet, List, Optional, Sequence + + +class Dialect(str, enum.Enum): + DUCKDB = "duckdb" + CLICKHOUSE = "clickhouse" + + +@dataclass(frozen=True) +class LogicalMeasure: + name: str + sql: str + aggregation: str = "SUM" + description: str = "" + + +@dataclass(frozen=True) +class LogicalDimension: + name: str + sql: str + description: str = "" + + +@dataclass(frozen=True) +class Role: + name: str + measures: FrozenSet[str] = field(default_factory=frozenset) + dimensions: FrozenSet[str] = field(default_factory=frozenset) + filters: FrozenSet[str] = field(default_factory=frozenset) + + +class SemanticModelError(Exception): + """Raised when a query violates the semantic contract or RBAC policy.""" + + +class SemanticModel: + """A formal semantic model: logical objects + role grants + dialect compile.""" + + def __init__(self) -> None: + self._measures: Dict[str, LogicalMeasure] = {} + self._dimensions: Dict[str, LogicalDimension] = {} + self._roles: Dict[str, Role] = {} + + # ------------------------------------------------------------------ # + # Registration + # ------------------------------------------------------------------ # + def add_measure(self, measure: LogicalMeasure) -> "SemanticModel": + self._measures[measure.name] = measure + return self + + def add_dimension(self, dimension: LogicalDimension) -> "SemanticModel": + self._dimensions[dimension.name] = dimension + return self + + def grant( + self, + name: str, + measures: Sequence[str] = (), + dimensions: Sequence[str] = (), + filters: Sequence[str] = (), + ) -> "SemanticModel": + self._roles[name] = Role( + name=name, + measures=frozenset(measures), + dimensions=frozenset(dimensions), + filters=frozenset(filters), + ) + return self + + # ------------------------------------------------------------------ # + # Validation + RBAC + # ------------------------------------------------------------------ # + def _require_role(self, role: str) -> Role: + r = self._roles.get(role) + if r is None: + raise SemanticModelError(f"Unknown role: {role}") + return r + + def authorize( + self, + role: str, + measures: Sequence[str], + dimensions: Sequence[str], + filters: Sequence[str] = (), + ) -> None: + """Raise :class:`SemanticModelError` if ``role`` cannot access the request.""" + r = self._require_role(role) + forbidden_measures = [m for m in measures if m not in r.measures] + if forbidden_measures: + raise SemanticModelError( + f"Role {role!r} may not access measures: {sorted(forbidden_measures)}" + ) + forbidden_dims = [d for d in dimensions if d not in r.dimensions] + if forbidden_dims: + raise SemanticModelError( + f"Role {role!r} may not access dimensions: {sorted(forbidden_dims)}" + ) + forbidden_filters = [f for f in filters if f not in r.filters] + if forbidden_filters: + raise SemanticModelError( + f"Role {role!r} may not filter on: {sorted(forbidden_filters)}" + ) + + # ------------------------------------------------------------------ # + # Dialect-aware compilation + # ------------------------------------------------------------------ # + @staticmethod + def _quote(identifier: str, dialect: Dialect) -> str: + if dialect is Dialect.CLICKHOUSE: + return f"`{identifier}`" + return f'"{identifier}"' + + def compile( + self, + role: str, + measures: Sequence[str], + dimensions: Sequence[str], + filters: Sequence[Dict[str, Any]] = (), + order_by: Optional[str] = None, + order_direction: str = "DESC", + limit: Optional[int] = None, + table: str = "telemetry", + dialect: Dialect = Dialect.DUCKDB, + ) -> str: + """Compile an authorized semantic query to dialect-specific SQL. + + Authorization is enforced here so no caller can bypass RBAC — if the + role is not entitled to a measure/dimension/filter the call raises. + """ + self.authorize( + role, + list(measures), + list(dimensions), + [f.get("field", "") for f in filters], + ) + + select: List[str] = [] + group_by: List[str] = [] + for m in measures: + measure = self._measures[m] + select.append( + f"{measure.aggregation}({measure.sql}) AS {self._quote(m, dialect)}" + ) + for d in dimensions: + dim = self._dimensions[d] + select.append(dim.sql) + group_by.append(dim.sql) + + parts = [f"SELECT {', '.join(select)}", f"FROM {self._quote(table, dialect)}"] + if group_by: + parts.append(f"GROUP BY {', '.join(group_by)}") + + where = [self._render_filter(f) for f in filters] + if where: + parts.append(f"WHERE {' AND '.join(where)}") + + if order_by: + measure = self._measures[order_by] + parts.append( + f"ORDER BY {measure.aggregation}({measure.sql}) {order_direction.upper()}" + ) + if limit: + parts.append(f"LIMIT {int(limit)}") + + return "\n".join(parts) + + @staticmethod + def _render_filter(filter_: Dict[str, Any]) -> str: + field = filter_["field"] + op = filter_.get("operator", "eq") + value = filter_.get("value") + if op == "eq": + return f"{field} = '{SemanticModel._esc(value)}'" + if op == "neq": + return f"{field} != '{SemanticModel._esc(value)}'" + if op in ("gt", "gte", "lt", "lte"): + return f"{field} {op} {value}" + if op == "between": + a, b = value + return f"{field} BETWEEN {a} AND {b}" + if op == "in": + items = ", ".join(f"'{SemanticModel._esc(v)}'" for v in value) + return f"{field} IN ({items})" + raise SemanticModelError(f"Unsupported operator: {op}") + + @staticmethod + def _esc(value: Any) -> str: + return str(value).replace("'", "''") + + # ------------------------------------------------------------------ # + # Compact prompt context (token-efficient semantic description) + # ------------------------------------------------------------------ # + def build_prompt_context( + self, role: str, dialect: Dialect = Dialect.DUCKDB + ) -> str: + """Return a concise, structured description of what ``role`` may query. + + This is fed to the LLM instead of the full DDL/schema, dramatically + shortening the prompt while still letting the model emit valid semantic + queries within the role's entitlements. + """ + r = self._require_role(role) + lines: List[str] = [f"Logical model (role={role}, engine={dialect.value}):"] + for m in sorted(r.measures): + measure = self._measures[m] + desc = f" — {measure.description}" if measure.description else "" + lines.append(f" measure: {m} ({measure.aggregation}){desc}") + for d in sorted(r.dimensions): + dim = self._dimensions[d] + desc = f" — {dim.description}" if dim.description else "" + lines.append(f" dimension: {d}{desc}") + if r.filters: + lines.append(f" filters: {', '.join(sorted(r.filters))}") + lines.append( + "Return JSON: {measures:[...], dimensions:[...], filters:[{field,operator,value}], order_by, limit}" + ) + return "\n".join(lines) + + +# --------------------------------------------------------------------------- # +# Default Gridify telemetry semantic model. +# +# Two roles are defined: ``analyst`` (full access) and ``viewer`` (a restricted +# role that can only see device-level aggregates, never raw score/uptime and +# never the ``status`` filter). This demonstrates role-based data security. +# --------------------------------------------------------------------------- # +def build_default_model() -> SemanticModel: + model = SemanticModel() + model.add_measure(LogicalMeasure("score", "score", "AVG", "device score")) + model.add_measure(LogicalMeasure("uptime", "uptime", "AVG", "uptime minutes")) + model.add_measure(LogicalMeasure("load_count", "load", "COUNT", "load events")) + model.add_measure(LogicalMeasure("temperature", "value", "AVG", "sensor temp")) + model.add_measure(LogicalMeasure("humidity", "value", "AVG", "sensor humidity")) + + model.add_dimension(LogicalDimension("device_id", "device_id", "device identifier")) + model.add_dimension(LogicalDimension("metric_type", "metric_type", "metric kind")) + model.add_dimension(LogicalDimension("status", "status", "device status")) + model.add_dimension(LogicalDimension("type", "type", "device type")) + model.add_dimension(LogicalDimension("timestamp", "timestamp", "event time")) + + model.grant( + "analyst", + measures=["score", "uptime", "load_count", "temperature", "humidity"], + dimensions=["device_id", "metric_type", "status", "type", "timestamp"], + filters=["device_id", "metric_type", "status", "type", "timestamp", "score", "uptime", "value"], + ) + model.grant( + "viewer", + measures=["temperature", "humidity", "load_count"], + dimensions=["device_id", "metric_type"], + filters=["device_id", "metric_type"], + ) + return model + + +_model: Optional[SemanticModel] = None + + +def get_semantic_model() -> SemanticModel: + global _model + if _model is None: + _model = build_default_model() + return _model diff --git a/backend/main.py b/backend/main.py index 3c3ce9d..07ec838 100644 --- a/backend/main.py +++ b/backend/main.py @@ -16,7 +16,7 @@ from fastapi.staticfiles import StaticFiles from app.config import settings -from app.routers import telemetry, gemini, olap +from app.routers import telemetry, gemini, olap, semantic from app.middleware.guardrail_middleware import AsyncGuardrailMiddleware from app.utils.logger import setup_logger @@ -55,6 +55,7 @@ async def lifespan(app: FastAPI): app.include_router(telemetry.router, prefix="/api", tags=["telemetry"]) app.include_router(gemini.router, prefix="/api", tags=["gemini"]) app.include_router(olap.router, prefix="/api", tags=["olap"]) +app.include_router(semantic.router, prefix="/api", tags=["semantic"]) # Async guardrails at the network boundary: prompt-injection scans run off the # synchronous request path (edge service or worker thread) so TTFT stays snappy. diff --git a/backend/tests/test_semantic_model.py b/backend/tests/test_semantic_model.py new file mode 100644 index 0000000..5b498a4 --- /dev/null +++ b/backend/tests/test_semantic_model.py @@ -0,0 +1,93 @@ +"""Tests for the formalized semantic model (RBAC, dialect compile, prompt context).""" + +import pytest + +from app.services.semantic_model import ( + Dialect, + LogicalDimension, + LogicalMeasure, + SemanticModel, + SemanticModelError, + build_default_model, + get_semantic_model, +) + + +def test_default_model_roles_compiled_sql(): + model = build_default_model() + sql = model.compile( + role="analyst", + measures=["temperature"], + dimensions=["device_id"], + limit=10, + ) + assert "AVG(value)" in sql + assert 'GROUP BY device_id' in sql or "GROUP BY device_id" in sql + assert "LIMIT 10" in sql + + +def test_viewer_cannot_access_restricted_measure(): + model = build_default_model() + with pytest.raises(SemanticModelError): + model.compile( + role="viewer", + measures=["score"], + dimensions=["device_id"], + ) + + +def test_viewer_cannot_filter_on_forbidden_field(): + model = build_default_model() + with pytest.raises(SemanticModelError): + model.compile( + role="viewer", + measures=["temperature"], + dimensions=["device_id"], + filters=[{"field": "status", "operator": "eq", "value": "alert"}], + ) + + +def test_authorize_helper_surfaces_forbidden_dimension(): + model = build_default_model() + with pytest.raises(SemanticModelError): + model.authorize("viewer", ["temperature"], ["status"]) + + +def test_clickhouse_dialect_quotes_identifiers(): + model = build_default_model() + sql = model.compile( + role="analyst", + measures=["temperature"], + dimensions=["device_id"], + dialect=Dialect.CLICKHOUSE, + ) + assert "`temperature`" in sql or "AVG(value)" in sql + assert sql.count("`") >= 2 # table + measure quoted + + +def test_prompt_context_is_compact_and_role_scoped(): + model = build_default_model() + ctx = model.build_prompt_context("viewer") + assert "role=viewer" in ctx + assert "score" not in ctx # analyst-only measure must not leak + assert "temperature" in ctx + assert "Return JSON" in ctx + + +def test_unknown_role_raises(): + model = build_default_model() + with pytest.raises(SemanticModelError): + model.build_prompt_context("root") + + +def test_factory_singleton(): + assert get_semantic_model() is get_semantic_model() + + +def test_register_custom_measure_and_grant(): + model = SemanticModel() + model.add_measure(LogicalMeasure("power", "watts", "SUM")) + model.add_dimension(LogicalDimension("site", "site_id")) + model.grant("ops", measures=["power"], dimensions=["site"]) + sql = model.compile("ops", ["power"], ["site"]) + assert "SUM(watts)" in sql From 3159f6f2154d642a889a99a30cc0e5b5a6dc9b22 Mon Sep 17 00:00:00 2001 From: Raymond Oyondi Date: Sun, 12 Jul 2026 19:05:57 +0000 Subject: [PATCH 3/6] Edge-RAG: int8 scalar quantization + chunked resumable model cache Cuts initialization cost ('Time to First Query') for ONNX/vector-index delivery on mobile and slow networks. - quantization.ts: float32 -> int8 scalar quantize/dequantize + int8 cosine - modelStreamCache.ts: chunked Cache Storage API delivery with resume + in-memory fallback for SSR/tests - ONNX worker returns quantized embeddings to shrink worker->main payload - RAG client caches/restores a quantized index so refreshes skip re-embed - Vitest coverage for quantization and chunked cache --- src/lib/modelStreamCache.test.ts | 64 ++++++++ src/lib/modelStreamCache.ts | 266 +++++++++++++++++++++++++++++++ src/lib/onnxWorker.ts | 46 +++++- src/lib/quantization.test.ts | 54 +++++++ src/lib/quantization.ts | 92 +++++++++++ src/lib/ragClient.ts | 33 +++- 6 files changed, 547 insertions(+), 8 deletions(-) create mode 100644 src/lib/modelStreamCache.test.ts create mode 100644 src/lib/modelStreamCache.ts create mode 100644 src/lib/quantization.test.ts create mode 100644 src/lib/quantization.ts diff --git a/src/lib/modelStreamCache.test.ts b/src/lib/modelStreamCache.test.ts new file mode 100644 index 0000000..ed0de36 --- /dev/null +++ b/src/lib/modelStreamCache.test.ts @@ -0,0 +1,64 @@ +import { describe, it, expect } from "vitest"; +import { + cacheModel, + loadModel, + cacheQuantizedIndex, + loadQuantizedIndex, + invalidateModel, + DEFAULT_CHUNK_SIZE, +} from "./modelStreamCache"; +import { dequantizeBatch } from "./quantization"; + +describe("chunked, resumable model cache", () => { + it("round-trips a model through chunked cache (in-memory fallback)", async () => { + const data = new Uint8Array(700 * 1024).map((_, i) => i % 251); + const manifest = await cacheModel("onnx-model", data.buffer.slice(0), { + cacheName: "test-cache", + }); + expect(manifest.totalChunks).toBeGreaterThan(1); + + const loaded = await loadModel("onnx-model", { cacheName: "test-cache" }); + expect(loaded).not.toBeNull(); + expect(loaded!.data.byteLength).toBe(data.length); + const got = new Uint8Array(loaded!.data); + expect(Array.from(got.slice(0, 8))).toEqual(Array.from(data.slice(0, 8))); + }); + + it("reports progress as chunks are fetched", async () => { + const data = new Uint8Array(DEFAULT_CHUNK_SIZE * 2 + 10); + await cacheModel("prog", data.buffer.slice(0), { cacheName: "test-cache" }); + const seen: number[] = []; + await loadModel("prog", { + cacheName: "test-cache", + onProgress: (fetched) => seen.push(fetched), + }); + expect(seen[seen.length - 1]).toBe(3); + }); + + it("persists and restores a quantized index", async () => { + const index = [ + [1, 2, 3], + [4, 5, 6], + ]; + await cacheQuantizedIndex("semantic-index", index, { cacheName: "test-cache" }); + const restored = await loadQuantizedIndex("semantic-index", { + cacheName: "test-cache", + }); + expect(restored).not.toBeNull(); + const deq = dequantizeBatch(restored!); + expect(deq).toHaveLength(2); + expect(deq[0][0]).toBeCloseTo(1, 1); + }); + + it("returns null when no manifest is cached", async () => { + const loaded = await loadModel("never-cached", { cacheName: "test-cache" }); + expect(loaded).toBeNull(); + }); + + it("invalidates a model completely", async () => { + const data = new Uint8Array(1024); + await cacheModel("to-del", data.buffer.slice(0), { cacheName: "test-cache" }); + await invalidateModel("to-del", { cacheName: "test-cache" }); + expect(await loadModel("to-del", { cacheName: "test-cache" })).toBeNull(); + }); +}); diff --git a/src/lib/modelStreamCache.ts b/src/lib/modelStreamCache.ts new file mode 100644 index 0000000..7f49b7d --- /dev/null +++ b/src/lib/modelStreamCache.ts @@ -0,0 +1,266 @@ +/** + * Chunked, resumable delivery of edge model + vector-index assets. + * + * Downloading the ONNX model and the full vector index to a mobile browser + * introduces a noticeable "Time to First Query" delay, and re-downloading them + * on every page refresh compounds it. This module: + * + * 1. Splits a payload into fixed-size chunks and persists them through the + * Cache Storage API (so they survive refreshes and are served offline). + * 2. Supports *resume*: only missing chunks are fetched, so a partial/broken + * cache is repaired incrementally rather than re-downloaded wholesale. + * 3. Falls back to an in-memory store when `caches` is unavailable (e.g. SSR + * or unit tests), keeping the same chunked contract. + * + * Pair this with `quantization.ts` (int8 embeddings) to shrink the index itself. + */ + +import { quantizeBatch, QuantizedVector } from "./quantization"; + +export const DEFAULT_CHUNK_SIZE = 256 * 1024; // 256 KiB + +export interface ModelManifest { + name: string; + totalChunks: number; + totalBytes: number; + chunkSize: number; + etag: string; + /** Optional quantized semantic index accompanying the model. */ + quantizedIndex?: QuantizedVector[]; +} + +// --------------------------------------------------------------------------- // +// Pluggable backing store +// --------------------------------------------------------------------------- // +interface ChunkStore { + put(key: string, value: ArrayBuffer): Promise; + get(key: string): Promise; + has(key: string): Promise; + delete(key: string): Promise; + keys(): Promise; +} + +/** Cache Storage API backing store (browser / worker). */ +class CacheStorageStore implements ChunkStore { + constructor(private readonly cacheName: string) {} + + private async _cache() { + if (typeof caches === "undefined") { + throw new Error("Cache Storage API unavailable"); + } + return caches.open(this.cacheName); + } + + async put(key: string, value: ArrayBuffer): Promise { + const cache = await this._cache(); + await cache.put( + key, + new Response(value, { + headers: { "Content-Type": "application/octet-stream" }, + }) + ); + } + + async get(key: string): Promise { + const cache = await this._cache(); + const res = await cache.match(key); + return res ? res.arrayBuffer() : null; + } + + async has(key: string): Promise { + return (await this.get(key)) !== null; + } + + async delete(key: string): Promise { + const cache = await this._cache(); + await cache.delete(key); + } + + async keys(): Promise { + const cache = await this._cache(); + return (await cache.keys()).map((req) => req.url); + } +} + +/** In-memory fallback store (Node / SSR / tests). */ +class MemoryStore implements ChunkStore { + private map = new Map(); + + async put(key: string, value: ArrayBuffer): Promise { + this.map.set(key, value); + } + async get(key: string): Promise { + return this.map.get(key) ?? null; + } + async has(key: string): Promise { + return this.map.has(key); + } + async delete(key: string): Promise { + this.map.delete(key); + } + async keys(): Promise { + return [...this.map.keys()]; + } +} + +function makeStore(cacheName: string): ChunkStore { + try { + if (typeof caches !== "undefined") return new CacheStorageStore(cacheName); + } catch { + /* fall through */ + } + return new MemoryStore(); +} + +// --------------------------------------------------------------------------- // +// Chunked model caching +// --------------------------------------------------------------------------- // +function chunkKey(name: string, index: number): string { + return `gridify:model:${name}:chunk:${index}`; +} +function manifestKey(name: string): string { + return `gridify:model:${name}:manifest`; +} + +function simpleEtag(data: ArrayBuffer): string { + let h = 2166136261; + const bytes = new Uint8Array(data); + const step = Math.max(1, Math.floor(bytes.length / 1024)); + for (let i = 0; i < bytes.length; i += step) { + h ^= bytes[i]; + h = Math.imul(h, 16777619); + } + return (h >>> 0).toString(16); +} + +/** Persist a model payload as resumable chunks. Returns the written manifest. */ +export async function cacheModel( + name: string, + data: ArrayBuffer, + opts: { cacheName?: string; chunkSize?: number } = {} +): Promise { + const chunkSize = opts.chunkSize ?? DEFAULT_CHUNK_SIZE; + const store = makeStore(opts.cacheName ?? "gridify-models"); + const totalChunks = Math.max(1, Math.ceil(data.byteLength / chunkSize)); + + const chunks: ArrayBuffer[] = []; + for (let i = 0; i < totalChunks; i++) { + const start = i * chunkSize; + chunks.push(data.slice(start, start + chunkSize)); + } + + const manifest: ModelManifest = { + name, + totalChunks, + totalBytes: data.byteLength, + chunkSize, + etag: simpleEtag(data), + }; + + for (let i = 0; i < totalChunks; i++) { + await store.put(chunkKey(name, i), chunks[i]); + } + await store.put(manifestKey(name), encodeJSON(manifest)); + return manifest; +} + +/** Cache a quantized semantic index alongside the model (no re-embed on reload). */ +export async function cacheQuantizedIndex( + name: string, + index: number[][], + opts: { cacheName?: string } = {} +): Promise { + const store = makeStore(opts.cacheName ?? "gridify-models"); + const quantized = quantizeBatch(index); + const manifest: ModelManifest = { + name, + totalChunks: 1, + totalBytes: index.length, + chunkSize: 1, + etag: String(index.length), + quantizedIndex: quantized, + }; + await store.put(manifestKey(name), encodeJSON(manifest)); + return manifest; +} + +/** + * Load a chunked model, resuming only the chunks that are missing from cache. + * Calls `onProgress` with the count of chunks fetched so far. Returns the + * reassembled payload, or `null` if no manifest is cached. + */ +export async function loadModel( + name: string, + opts: { + cacheName?: string; + onProgress?: (fetched: number, total: number) => void; + fetchChunk?: ( + name: string, + index: number, + manifest: ModelManifest + ) => Promise; + } = {} +): Promise<{ data: ArrayBuffer; manifest: ModelManifest } | null> { + const store = makeStore(opts.cacheName ?? "gridify-models"); + const manifestBuf = await store.get(manifestKey(name)); + if (!manifestBuf) return null; + const manifest = decodeJSON(manifestBuf); + + const out = new Uint8Array(manifest.totalBytes); + let fetched = 0; + for (let i = 0; i < manifest.totalChunks; i++) { + let chunk = await store.get(chunkKey(name, i)); + if (!chunk) { + // Hook for a remote fetch; callers may override `fetchChunk`. + chunk = (await opts.fetchChunk?.(name, i, manifest)) ?? null; + if (!chunk) throw new Error(`Missing chunk ${i} for model ${name}`); + await store.put(chunkKey(name, i), chunk); + } + out.set(new Uint8Array(chunk), i * manifest.chunkSize); + fetched++; + opts.onProgress?.(fetched, manifest.totalChunks); + } + return { data: out.buffer, manifest }; +} + +/** Read a previously cached quantized index, if present. */ +export async function loadQuantizedIndex( + name: string, + opts: { cacheName?: string } = {} +): Promise { + const store = makeStore(opts.cacheName ?? "gridify-models"); + const manifestBuf = await store.get(manifestKey(name)); + if (!manifestBuf) return null; + const manifest = decodeJSON(manifestBuf); + return manifest.quantizedIndex ?? null; +} + +/** Clear one model's chunks + manifest from the cache. */ +export async function invalidateModel( + name: string, + opts: { cacheName?: string } = {} +): Promise { + const store = makeStore(opts.cacheName ?? "gridify-models"); + const manifestBuf = await store.get(manifestKey(name)); + if (manifestBuf) { + const manifest = decodeJSON(manifestBuf); + for (let i = 0; i < manifest.totalChunks; i++) { + await store.delete(chunkKey(name, i)); + } + } + await store.delete(manifestKey(name)); +} + +function encodeJSON(value: unknown): ArrayBuffer { + const text = JSON.stringify(value); + const buf = new Uint8Array(text.length); + for (let i = 0; i < text.length; i++) buf[i] = text.charCodeAt(i); + return buf.buffer; +} + +function decodeJSON(buf: ArrayBuffer): T { + const bytes = new Uint8Array(buf); + let text = ""; + for (let i = 0; i < bytes.length; i++) text += String.fromCharCode(bytes[i]); + return JSON.parse(text) as T; +} diff --git a/src/lib/onnxWorker.ts b/src/lib/onnxWorker.ts index 2b843ac..714ab17 100644 --- a/src/lib/onnxWorker.ts +++ b/src/lib/onnxWorker.ts @@ -14,6 +14,26 @@ async function init(modelUrl: string, modelDim = 64): Promise { outputName = session.outputNames[0]; } +/** Quantize a float32 embedding to int8 to shrink the worker -> main payload. */ +function quantize( + vec: ArrayLike +): { codes: number[]; scale: number; min: number } { + let min = Infinity; + let max = -Infinity; + for (let i = 0; i < vec.length; i++) { + const v = vec[i]; + if (v < min) min = v; + if (v > max) max = v; + } + const scale = max - min > 0 ? (max - min) / 255 : 1; + const codes: number[] = []; + for (let i = 0; i < vec.length; i++) { + const q = Math.round((vec[i] - min) / scale); + codes.push(Math.max(0, Math.min(255, q)) - 128); + } + return { codes, scale, min }; +} + function hashEmbedding(text: string, modelDim: number): number[] { const vec = new Array(modelDim).fill(0); const tokens = text.toLowerCase().match(/[a-z0-9]+/g) ?? []; @@ -30,11 +50,15 @@ function hashEmbedding(text: string, modelDim: number): number[] { return vec; } -async function embed(texts: string[]): Promise { +async function embed( + texts: string[], + quantize = false +): Promise<{ embeddings: number[][]; quantized?: ReturnType[] }> { if (!session) { throw new Error("ONNX worker not initialized"); } const results: number[][] = []; + const quantized: ReturnType[] = []; for (const text of texts) { const vec = hashEmbedding(text, dim); const tensor = { @@ -43,10 +67,12 @@ async function embed(texts: string[]): Promise { type: "float32" as const, } as any; const out = await session.run({ [inputName]: tensor }); - const data = out[outputName].data; - results.push(Array.from(data as Float32Array)); + const data = out[outputName].data as Float32Array; + const full = Array.from(data); + results.push(full); + if (quantize) quantized.push(quantize(full)); } - return results; + return { embeddings: results, quantized: quantize ? quantized : undefined }; } export {}; @@ -57,15 +83,21 @@ self.onmessage = async (e: MessageEvent<{ modelUrl?: string; modelDim?: number; texts?: string[]; + quantize?: boolean; }>) => { - const { type, requestId, modelUrl, modelDim, texts } = e.data; + const { type, requestId, modelUrl, modelDim, texts, quantize } = e.data; try { if (type === "init") { await init(modelUrl!, modelDim ?? 64); self.postMessage({ type: "ready", requestId }); } else if (type === "embed") { - const embeddings = await embed(texts!); - self.postMessage({ type: "result", requestId, embeddings }); + const { embeddings, quantized } = await embed(texts!, quantize); + self.postMessage({ + type: "result", + requestId, + embeddings, + quantized, + }); } } catch (err) { self.postMessage({ diff --git a/src/lib/quantization.test.ts b/src/lib/quantization.test.ts new file mode 100644 index 0000000..8d64434 --- /dev/null +++ b/src/lib/quantization.test.ts @@ -0,0 +1,54 @@ +import { describe, it, expect } from "vitest"; +import { + quantizeVector, + dequantizeVector, + quantizeBatch, + dequantizeBatch, + quantizedCosine, + storageEstimate, +} from "./quantization"; + +describe("scalar quantization (float32 -> int8)", () => { + it("round-trips a vector with small error", () => { + const v = [0.1, -0.5, 0.9, 0.0, 0.42]; + const q = quantizeVector(v); + const back = dequantizeVector(q); + expect(back.length).toBe(v.length); + for (let i = 0; i < v.length; i++) { + expect(Math.abs(back[i] - v[i])).toBeLessThan(0.05); + } + }); + + it("produces codes within the int8 range", () => { + const q = quantizeVector([0, 1, 2, 3, 4, 5]); + for (const c of q.codes) { + expect(c).toBeGreaterThanOrEqual(-128); + expect(c).toBeLessThanOrEqual(127); + } + }); + + it("batch quantize/dequantize preserves length and shape", () => { + const vectors = [ + [1, 2, 3], + [4, 5, 6], + ]; + const back = dequantizeBatch(quantizeBatch(vectors)); + expect(back).toHaveLength(2); + expect(back[0]).toHaveLength(3); + }); + + it("cosine on int8 codes matches float cosine for similar vectors", () => { + const a = quantizeVector([1, 0, 0]); + const b = quantizeVector([1, 0.1, 0]); + const c = quantizeVector([0, 1, 0]); + expect(quantizedCosine(a, b)).toBeGreaterThan(quantizedCosine(a, c)); + }); + + it("reports a ~4x storage reduction", () => { + const est = storageEstimate( + Array.from({ length: 100 }, () => new Array(64).fill(0)) + ); + expect(est.ratio).toBeCloseTo(4, 1); + expect(est.int8Bytes * 4).toBe(est.float32Bytes); + }); +}); diff --git a/src/lib/quantization.ts b/src/lib/quantization.ts new file mode 100644 index 0000000..091b318 --- /dev/null +++ b/src/lib/quantization.ts @@ -0,0 +1,92 @@ +/** + * Scalar quantization for edge-RAG embeddings. + * + * ONNX Runtime Web runs semantic pre-filtering on the client, but shipping + * full `float32` embeddings for every chunk of the vector index is expensive + * on mobile / slow networks ("Time to First Query"). We compress each vector to + * `int8` with a per-vector scale (and shared zero-point 0), cutting payload and + * cache size by ~4x with negligible loss for cosine pre-filtering. + * + * Dequantization restores an approximation of the original float32 vector so the + * existing cosine-similarity math in `ragClient` is unchanged. + */ + +export interface QuantizedVector { + /** int8 coefficients, each in [0, 255]. */ + codes: Int8Array; + /** scale used to map float32 -> int8. */ + scale: number; + /** original minimum value (zero-point is 0). */ + min: number; +} + +/** Quantize a single float32 vector to int8 (scalar / uniform quantization). */ +export function quantizeVector(vec: ArrayLike): QuantizedVector { + let min = Infinity; + let max = -Infinity; + for (let i = 0; i < vec.length; i++) { + const v = vec[i]; + if (v < min) min = v; + if (v > max) max = v; + } + const range = max - min; + const scale = range > 0 ? range / 255 : 1; + const codes = new Int8Array(vec.length); + for (let i = 0; i < vec.length; i++) { + const q = Math.round((vec[i] - min) / scale); + codes[i] = Math.max(0, Math.min(255, q)) - 128; + } + return { codes, scale, min }; +} + +/** Restore an approximate float32 vector from its int8 codes. */ +export function dequantizeVector(q: QuantizedVector): Float32Array { + const out = new Float32Array(q.codes.length); + for (let i = 0; i < q.codes.length; i++) { + out[i] = (q.codes[i] + 128) * q.scale + q.min; + } + return out; +} + +/** Quantize a batch of vectors (e.g. a full semantic index). */ +export function quantizeBatch(vectors: number[][]): QuantizedVector[] { + return vectors.map(quantizeVector); +} + +/** Dequantize a batch of vectors. */ +export function dequantizeBatch(vectors: QuantizedVector[]): number[][] { + return vectors.map((v) => Array.from(dequantizeVector(v))); +} + +/** Cosine similarity computed directly on int8 codes (avoids dequant round-trip). */ +export function quantizedCosine(a: QuantizedVector, b: QuantizedVector): number { + const n = Math.min(a.codes.length, b.codes.length); + if (n === 0) return 0; + let dot = 0; + let na = 0; + let nb = 0; + for (let i = 0; i < n; i++) { + const x = a.codes[i]; + const y = b.codes[i]; + dot += x * y; + na += x * x; + nb += y * y; + } + if (na === 0 || nb === 0) return 0; + return dot / (Math.sqrt(na) * Math.sqrt(nb)); +} + +/** + * Estimate the storage reduction of int8 vs float32 for a batch. + * Returns the byte size before and after quantization. + */ +export function storageEstimate(vectors: number[][]): { + float32Bytes: number; + int8Bytes: number; + ratio: number; +} { + const float32Bytes = vectors.length * (vectors[0]?.length ?? 0) * 4; + const int8Bytes = vectors.length * (vectors[0]?.length ?? 0); + const ratio = float32Bytes > 0 ? float32Bytes / int8Bytes : 1; + return { float32Bytes, int8Bytes, ratio }; +} diff --git a/src/lib/ragClient.ts b/src/lib/ragClient.ts index 8a405e8..db5ad94 100644 --- a/src/lib/ragClient.ts +++ b/src/lib/ragClient.ts @@ -16,6 +16,8 @@ * block the main UI thread. */ +import { cacheQuantizedIndex, dequantizeBatch, loadQuantizedIndex } from "./modelStreamCache"; + export interface RAGChunk { id: string; text: string; @@ -208,9 +210,25 @@ export class HybridRAG { })); } - /** Load the cached semantic index; falls back to local docs on any failure. */ + /** + * Load the cached semantic index. On a warm cache we restore the *quantized* + * index (int8) from the Cache Storage API so no ONNX re-embedding happens on + * page refresh. On a cold cache we build/load embeddings and persist a + * quantized copy for next time. + */ async loadIndex(remoteUrl?: string): Promise { if (this.index.length > 0) return; + + const cached = await loadQuantizedIndex("semantic-index"); + if (cached) { + const embeddings = dequantizeBatch(cached); + this.index = LOCAL_DOCUMENTS.map((d, i) => ({ + ...d, + embedding: embeddings[i] ?? hashEmbedding(d.text, this.embedder.dim), + })); + return; + } + if (remoteUrl) { try { const res = await fetch(remoteUrl, { headers: { Accept: "application/json" } }); @@ -226,6 +244,7 @@ export class HybridRAG { } } this.index = payload.chunks; + await this._persistQuantizedIndex(); return; } } @@ -234,6 +253,18 @@ export class HybridRAG { } } await this._buildFromLocal(); + await this._persistQuantizedIndex(); + } + + /** Quantize the current index to int8 and cache it for resumable reloads. */ + private async _persistQuantizedIndex(): Promise { + if (this.index.length === 0) return; + const vectors = this.index.map((c) => c.embedding ?? hashEmbedding(c.text, this.embedder.dim)); + try { + await cacheQuantizedIndex("semantic-index", vectors); + } catch { + // Cache Storage unavailable (e.g. SSR) — ignore; reload will rebuild. + } } /** Local cosine-match. Returns the top-k chunks for the query. */ From 9cc6bf85fca996148bab29920747f8f69aa1b680 Mon Sep 17 00:00:00 2001 From: Raymond Oyondi Date: Sun, 12 Jul 2026 19:09:12 +0000 Subject: [PATCH 4/6] Real-time collaborative layout sync (CRDT provider + persistence) GenAI drafts and Framer-Motion drags now stream to every active session instead of living only in ephemeral Zustand state. - realtimeSync.ts: provider abstraction (BroadcastChannel default, Yjs multiplayer CRDT lazy-loaded, Supabase hosted) + DashboardSync binder - LayoutRepository: in-memory default + PostgreSQL store for durable, shareable, cross-pod layout persistence - /api/layouts CRUD (save/load/list/delete, public sharing) - REALTIME_PROVIDER + Yjs/Supabase config - Vitest + pytest coverage for sync convergence and layout storage --- backend/app/config.py | 9 + backend/app/routers/layouts.py | 62 +++++ backend/app/services/layout_repository.py | 231 ++++++++++++++++ backend/main.py | 3 +- backend/tests/test_layout_repository.py | 70 +++++ src/lib/realtimeSync.test.ts | 89 ++++++ src/lib/realtimeSync.ts | 314 ++++++++++++++++++++++ 7 files changed, 777 insertions(+), 1 deletion(-) create mode 100644 backend/app/routers/layouts.py create mode 100644 backend/app/services/layout_repository.py create mode 100644 backend/tests/test_layout_repository.py create mode 100644 src/lib/realtimeSync.test.ts create mode 100644 src/lib/realtimeSync.ts diff --git a/backend/app/config.py b/backend/app/config.py index 882d600..8c625d9 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -75,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" diff --git a/backend/app/routers/layouts.py b/backend/app/routers/layouts.py new file mode 100644 index 0000000..8ff4952 --- /dev/null +++ b/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} diff --git a/backend/app/services/layout_repository.py b/backend/app/services/layout_repository.py new file mode 100644 index 0000000..735b636 --- /dev/null +++ b/backend/app/services/layout_repository.py @@ -0,0 +1,231 @@ +"""Persistent storage for saved / shared dashboard layouts. + +GenAI drafts and dragged widgets live in Zustand for ephemeral UI state, but +users need to *save*, *share*, and *collaborate* on layouts. This repository +persists layouts so they survive reloads and can be opened by others. The +in-memory implementation is the default (zero infra); when a PostgreSQL +connection is available the same interface is backed by a real table so saved +layouts are durable and queryable across pods. +""" + +from __future__ import annotations + +import json +import uuid +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Dict, List, Optional + + +@dataclass +class DashboardLayout: + id: str + name: str + owner: str + widgets: List[dict] = field(default_factory=list) + is_public: bool = False + created_at: str = "" + updated_at: str = "" + + def to_dict(self) -> Dict: + return { + "id": self.id, + "name": self.name, + "owner": self.owner, + "widgets": self.widgets, + "is_public": self.is_public, + "created_at": self.created_at, + "updated_at": self.updated_at, + } + + +class LayoutRepository: + """Interface for layout persistence.""" + + def save(self, layout: DashboardLayout) -> DashboardLayout: + raise NotImplementedError + + def get(self, layout_id: str) -> Optional[DashboardLayout]: + raise NotImplementedError + + def list_by_owner(self, owner: str) -> List[DashboardLayout]: + raise NotImplementedError + + def list_public(self) -> List[DashboardLayout]: + raise NotImplementedError + + def delete(self, layout_id: str) -> bool: + raise NotImplementedError + + +class InMemoryLayoutRepository(LayoutRepository): + """Default zero-infra repository (also used in tests).""" + + def __init__(self) -> None: + self._store: Dict[str, DashboardLayout] = {} + + def save(self, layout: DashboardLayout) -> DashboardLayout: + now = datetime.now(timezone.utc).isoformat() + existing = self._store.get(layout.id) + layout.created_at = existing.created_at or now + layout.updated_at = now + self._store[layout.id] = layout + return layout + + def get(self, layout_id: str) -> Optional[DashboardLayout]: + return self._store.get(layout_id) + + def list_by_owner(self, owner: str) -> List[DashboardLayout]: + return [l for l in self._store.values() if l.owner == owner] + + def list_public(self) -> List[DashboardLayout]: + return [l for l in self._store.values() if l.is_public] + + def delete(self, layout_id: str) -> bool: + return self._store.pop(layout_id, None) is not None + + +class PostgresLayoutRepository(LayoutRepository): + """Durable, cross-pod repository backed by PostgreSQL via SQLAlchemy Core.""" + + def __init__(self, database_url: str): + from sqlalchemy import create_engine, text + + self._engine = create_engine(database_url, pool_pre_ping=True) + self._text = text + self._ensure_schema() + + def _ensure_schema(self) -> None: + with self._engine.begin() as conn: + conn.execute( + self._text( + """ + CREATE TABLE IF NOT EXISTS dashboard_layouts ( + id VARCHAR PRIMARY KEY, + name VARCHAR NOT NULL, + owner VARCHAR NOT NULL, + widgets JSON NOT NULL, + is_public BOOLEAN NOT NULL DEFAULT FALSE, + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL + ) + """ + ) + ) + + def _row_to_layout(self, row) -> DashboardLayout: + return DashboardLayout( + id=row.id, + name=row.name, + owner=row.owner, + widgets=json.loads(row.widgets) if isinstance(row.widgets, str) else row.widgets, + is_public=row.is_public, + created_at=row.created_at.isoformat() if row.created_at else "", + updated_at=row.updated_at.isoformat() if row.updated_at else "", + ) + + def save(self, layout: DashboardLayout) -> DashboardLayout: + from sqlalchemy import text + + now = datetime.now(timezone.utc) + existing = self.get(layout.id) + created = existing.created_at or now.isoformat() + with self._engine.begin() as conn: + conn.execute( + text( + """ + INSERT INTO dashboard_layouts + (id, name, owner, widgets, is_public, created_at, updated_at) + VALUES (:id, :name, :owner, :widgets, :public, :created, :updated) + ON CONFLICT (id) DO UPDATE SET + name = EXCLUDED.name, + owner = EXCLUDED.owner, + widgets = EXCLUDED.widgets, + is_public = EXCLUDED.is_public, + updated_at = EXCLUDED.updated_at + """ + ), + { + "id": layout.id, + "name": layout.name, + "owner": layout.owner, + "widgets": json.dumps(layout.widgets), + "public": layout.is_public, + "created": existing.created_at or now, + "updated": now, + }, + ) + layout.created_at = created + layout.updated_at = now.isoformat() + return layout + + def get(self, layout_id: str) -> Optional[DashboardLayout]: + from sqlalchemy import text + + with self._engine.connect() as conn: + row = conn.execute( + text("SELECT * FROM dashboard_layouts WHERE id = :id"), + {"id": layout_id}, + ).fetchone() + return self._row_to_layout(row) if row else None + + def list_by_owner(self, owner: str) -> List[DashboardLayout]: + from sqlalchemy import text + + with self._engine.connect() as conn: + rows = conn.execute( + text("SELECT * FROM dashboard_layouts WHERE owner = :owner"), + {"owner": owner}, + ).fetchall() + return [self._row_to_layout(r) for r in rows] + + def list_public(self) -> List[DashboardLayout]: + from sqlalchemy import text + + with self._engine.connect() as conn: + rows = conn.execute( + text("SELECT * FROM dashboard_layouts WHERE is_public = TRUE") + ).fetchall() + return [self._row_to_layout(r) for r in rows] + + def delete(self, layout_id: str) -> bool: + from sqlalchemy import text + + with self._engine.begin() as conn: + res = conn.execute( + text("DELETE FROM dashboard_layouts WHERE id = :id"), + {"id": layout_id}, + ) + return res.rowcount > 0 + + +_repo: Optional[LayoutRepository] = None + + +def get_layout_repository() -> LayoutRepository: + """Return the active layout repository. + + Uses PostgreSQL when ``DATABASE_URL`` is configured and reachable, otherwise + the in-memory store. This keeps the app runnable with zero infra while still + persisting layouts in production. + """ + global _repo + if _repo is not None: + return _repo + from app.config import settings + + if settings.DATABASE_URL: + try: + _repo = PostgresLayoutRepository(settings.DATABASE_URL) + return _repo + except Exception: + # Fall back to in-memory when Postgres is unreachable. + pass + _repo = InMemoryLayoutRepository() + return _repo + + +def reset_layout_repository(instance: Optional[LayoutRepository] = None) -> None: + """Replace the cached repository (used in tests).""" + global _repo + _repo = instance diff --git a/backend/main.py b/backend/main.py index 07ec838..e098f7d 100644 --- a/backend/main.py +++ b/backend/main.py @@ -16,7 +16,7 @@ from fastapi.staticfiles import StaticFiles from app.config import settings -from app.routers import telemetry, gemini, olap, semantic +from app.routers import telemetry, gemini, olap, semantic, layouts from app.middleware.guardrail_middleware import AsyncGuardrailMiddleware from app.utils.logger import setup_logger @@ -56,6 +56,7 @@ async def lifespan(app: FastAPI): app.include_router(gemini.router, prefix="/api", tags=["gemini"]) app.include_router(olap.router, prefix="/api", tags=["olap"]) app.include_router(semantic.router, prefix="/api", tags=["semantic"]) +app.include_router(layouts.router, prefix="/api", tags=["layouts"]) # Async guardrails at the network boundary: prompt-injection scans run off the # synchronous request path (edge service or worker thread) so TTFT stays snappy. diff --git a/backend/tests/test_layout_repository.py b/backend/tests/test_layout_repository.py new file mode 100644 index 0000000..6edd0e7 --- /dev/null +++ b/backend/tests/test_layout_repository.py @@ -0,0 +1,70 @@ +"""Tests for persistent dashboard-layout storage.""" + +import pytest + +from app.services.layout_repository import ( + DashboardLayout, + InMemoryLayoutRepository, + get_layout_repository, + reset_layout_repository, +) + + +@pytest.fixture() +def repo(): + r = InMemoryLayoutRepository() + reset_layout_repository(r) + yield r + reset_layout_repository(None) + + +def _layout(owner="u1", public=False): + return DashboardLayout( + id="L1", + name="My Layout", + owner=owner, + widgets=[{"id": "w1", "title": "T"}], + is_public=public, + ) + + +def test_save_and_get(repo): + saved = repo.save(_layout()) + assert saved.id == "L1" + assert saved.created_at and saved.updated_at + got = repo.get("L1") + assert got is not None + assert got.name == "My Layout" + assert got.widgets == [{"id": "w1", "title": "T"}] + + +def test_overwrite_updates_timestamp(repo): + first = repo.save(_layout()) + later = repo.save(_layout()) + assert later.created_at == first.created_at # preserved + assert later.updated_at >= first.updated_at + + +def test_list_by_owner_and_public(repo): + repo.save(_layout(owner="u1", public=False)) + repo.save(_layout(owner="u2", public=True)) + repo.save(_layout(owner="u2", public=False)) + assert len(repo.list_by_owner("u2")) == 2 + assert len(repo.list_public()) == 1 + + +def test_delete(repo): + repo.save(_layout()) + assert repo.delete("L1") is True + assert repo.get("L1") is None + assert repo.delete("L1") is False + + +def test_factory_defaults_to_in_memory(monkeypatch): + monkeypatch.setattr("app.config.settings.DATABASE_URL", "") + reset_layout_repository(None) + repo = get_layout_repository() + try: + assert isinstance(repo, InMemoryLayoutRepository) + finally: + reset_layout_repository(None) diff --git a/src/lib/realtimeSync.test.ts b/src/lib/realtimeSync.test.ts new file mode 100644 index 0000000..ea3d33a --- /dev/null +++ b/src/lib/realtimeSync.test.ts @@ -0,0 +1,89 @@ +import { describe, it, expect } from "vitest"; +import { + BroadcastChannelProvider, + DashboardSync, + createProvider, + RealtimeProvider, + SyncMessage, +} from "./realtimeSync"; +import { Widget } from "../types"; + +const widgetsA: Widget[] = [ + { id: "a", title: "A", subtitle: "", type: "line", w: 4, order: 0 }, +]; +const widgetsB: Widget[] = [ + { id: "b", title: "B", subtitle: "", type: "bar", w: 4, order: 0 }, +]; + +describe("BroadcastChannelProvider + DashboardSync", () => { + it("streams local layout changes to a remote session", async () => { + const received: Widget[][] = []; + const remote = new DashboardSync({ + provider: new BroadcastChannelProvider(), + room: "room-1", + initial: widgetsA, + onChange: (w) => received.push(w), + }); + await remote.start(); + + const local = new DashboardSync({ + provider: new BroadcastChannelProvider(), + room: "room-1", + initial: widgetsA, + onChange: () => {}, + }); + await local.start(); + local.update(widgetsB); + + // Give the in-process hub a tick to deliver. + await new Promise((r) => setTimeout(r, 0)); + + expect(received.length).toBeGreaterThanOrEqual(1); + expect(received[received.length - 1][0].id).toBe("b"); + remote.stop(); + local.stop(); + }); + + it("provider delivers both messages (staleness handled by DashboardSync)", async () => { + const received: SyncMessage[] = []; + const provider = new BroadcastChannelProvider(); + await provider.connect("room-2"); + provider.onMessage((m) => received.push(m)); + + provider.broadcast({ kind: "update", widgets: widgetsB, revision: 100 }); + provider.broadcast({ kind: "update", widgets: widgetsA, revision: 50 }); + + await new Promise((r) => setTimeout(r, 0)); + expect(received).toHaveLength(2); + provider.disconnect(); + }); + + it("createProvider returns the requested transport", () => { + expect(createProvider("broadcast").name).toBe("broadcast"); + const y = createProvider("yjs"); + expect((y as RealtimeProvider).name).toBe("yjs"); + }); + + it("two DashboardSync instances converge on the same room", async () => { + const changes: Widget[][] = []; + const a = new DashboardSync({ + provider: new BroadcastChannelProvider(), + room: "room-3", + initial: widgetsA, + onChange: () => {}, + }); + const b = new DashboardSync({ + provider: new BroadcastChannelProvider(), + room: "room-3", + initial: widgetsA, + onChange: (w) => changes.push(w), + }); + await a.start(); + await b.start(); + a.update(widgetsB); + await new Promise((r) => setTimeout(r, 0)); + expect(changes.some((c) => c[0]?.id === "b")).toBe(true); + a.stop(); + b.stop(); + }); +}); diff --git a/src/lib/realtimeSync.ts b/src/lib/realtimeSync.ts new file mode 100644 index 0000000..a60d5ad --- /dev/null +++ b/src/lib/realtimeSync.ts @@ -0,0 +1,314 @@ +/** + * Real-time sync engine for collaborative dashboard layouts. + * + * GenAI drafts and Framer-Motion drags change dashboard layouts constantly. + * Zustand holds ephemeral UI state, but saved/sharable layouts need to stream + * to every active browser session the moment they change. This module provides + * a provider abstraction so the transport can be swapped without touching the + * dashboard code: + * + * - `BroadcastChannelProvider` : zero-dep, same-browser cross-tab sync + * (falls back to an in-process hub when `BroadcastChannel` is unavailable, + * e.g. unit tests / SSR). Good enough for single-device multi-window. + * - `YjsProvider` : multiplayer CRDT over a WebSocket + * (Yjs + y-websocket), the recommended option for true cross-device, + * conflict-free collaborative editing. Loaded lazily so the bundle and the + * test environment never require Yjs unless this provider is selected. + * - `SupabaseProvider` : hosted realtime (Supabase Realtime / Postgres + * changes) — a thin wrapper illustrating the hosted alternative. + * + * `DashboardSync` ties a provider to a layout document: local edits are pushed + * immediately and remote edits are applied through a single `onChange` callback. + */ + +import { Widget } from "../types"; + +export type SyncKind = "update" | "init" | "presence"; + +export interface SyncMessage { + kind: SyncKind; + room: string; + sender: string; + widgets: Widget[]; + /** Wall-clock ts used to break ties on concurrent updates. */ + revision: number; +} + +export interface RealtimeProvider { + readonly name: string; + connect(room: string): Promise; + broadcast(message: Omit): void; + onMessage(cb: (msg: SyncMessage) => void): void; + disconnect(): void; +} + +// --------------------------------------------------------------------------- // +// In-process hub: lets multiple provider instances in the same JS realm (tests, +// SSR) exchange messages when the platform `BroadcastChannel` is missing. +// --------------------------------------------------------------------------- // +const __hubs = new Map void>>(); + +function hubPublish(room: string, msg: SyncMessage): void { + const subs = __hubs.get(room); + if (!subs) return; + for (const cb of subs) cb(msg); +} +function hubSubscribe(room: string, cb: (msg: SyncMessage) => void): () => void { + let subs = __hubs.get(room); + if (!subs) { + subs = new Set(); + __hubs.set(room, subs); + } + subs.add(cb); + return () => subs!.delete(cb); +} + +let __senderId = 0; +function nextSender(): string { + __senderId += 1; + return `client-${__senderId}-${Math.random().toString(36).slice(2, 8)}`; +} + +// --------------------------------------------------------------------------- // +// Provider: cross-tab via BroadcastChannel (or in-process hub fallback) +// --------------------------------------------------------------------------- // +export class BroadcastChannelProvider implements RealtimeProvider { + readonly name = "broadcast"; + private room = ""; + private channel: BroadcastChannel | null = null; + private sender = nextSender(); + private listeners = new Set<(msg: SyncMessage) => void>(); + + async connect(room: string): Promise { + this.room = room; + if (typeof BroadcastChannel !== "undefined") { + this.channel = new BroadcastChannel(`gridify:${room}`); + this.channel.onmessage = (e: MessageEvent) => { + if (e.data.sender !== this.sender) { + for (const cb of this.listeners) cb(e.data); + } + }; + } + } + + broadcast(message: Omit): void { + const full: SyncMessage = { ...message, room: this.room, sender: this.sender }; + if (this.channel) { + this.channel.postMessage(full); + } else { + hubPublish(this.room, full); + } + } + + onMessage(cb: (msg: SyncMessage) => void): void { + if (this.channel) { + this.listeners.add(cb); + } else { + this._hubUnsub = hubSubscribe(this.room, (msg) => { + if (msg.sender !== this.sender) cb(msg); + }); + } + } + + private _hubUnsub: (() => void) | null = null; + + disconnect(): void { + this._hubUnsub?.(); + this._hubUnsub = null; + this.listeners.clear(); + this.channel?.close(); + this.channel = null; + } +} + +// --------------------------------------------------------------------------- // +// Provider: Yjs multiplayer CRDT (lazy import — no bundle/test cost unless used) +// --------------------------------------------------------------------------- // +export class YjsProvider implements RealtimeProvider { + readonly name = "yjs"; + private room = ""; + private sender = nextSender(); + private doc: any = null; + private array: any = null; + private listeners = new Set<(msg: SyncMessage) => void>(); + private wsUrl: string; + + constructor(opts: { wsUrl?: string } = {}) { + this.wsUrl = opts.wsUrl ?? "wss://gridify.yjs.dev"; + } + + async connect(room: string): Promise { + this.room = room; + let Y: any; + let WebsocketProvider: any; + try { + Y = await import("yjs"); + ({ WebsocketProvider } = await import("y-websocket")); + } catch (err) { + throw new Error( + "YjsProvider requires 'yjs' and 'y-websocket' to be installed. " + + "Add them as dependencies to enable multiplayer CRDT sync." + ); + } + this.doc = new Y.Doc(); + this.array = this.doc.getArray("widgets"); + new WebsocketProvider(this.wsUrl, room, this.doc); + this.array.observe(() => { + const msg: SyncMessage = { + kind: "update", + room: this.room, + sender: this.sender, + widgets: this.array.toArray() as Widget[], + revision: Date.now(), + }; + for (const cb of this.listeners) cb(msg); + }); + } + + broadcast(message: Omit): void { + if (!this.array) return; + this.array.delete(0, this.array.length); + this.array.insert(0, message.widgets); + } + + onMessage(cb: (msg: SyncMessage) => void): void { + this.listeners.add(cb); + } + + disconnect(): void { + this.listeners.clear(); + try { + this.doc?.destroy(); + } catch { + /* ignore */ + } + } +} + +// --------------------------------------------------------------------------- // +// Provider: Supabase hosted realtime (illustrative thin wrapper) +// --------------------------------------------------------------------------- // +export class SupabaseProvider implements RealtimeProvider { + readonly name = "supabase"; + private room = ""; + private sender = nextSender(); + private listeners = new Set<(msg: SyncMessage) => void>(); + private client: any = null; + + constructor(opts: { url?: string; anonKey?: string } = {}) { + this.url = opts.url; + this.anonKey = opts.anonKey; + } + private url?: string; + private anonKey?: string; + + async connect(room: string): Promise { + this.room = room; + if (!this.url) return; // no-op when unconfigured + const mod = await import("@supabase/supabase-js").catch(() => null); + if (!mod) return; + this.client = mod.createClient(this.url, this.anonKey); + this.client + .channel(`layout:${room}`) + .on("broadcast", { event: "sync" }, (payload: { payload: SyncMessage }) => { + if (payload.payload.sender !== this.sender) { + for (const cb of this.listeners) cb(payload.payload); + } + }) + .subscribe(); + } + + broadcast(message: Omit): void { + if (!this.client) return; + this.client.channel(`layout:${this.room}`).send({ + type: "broadcast", + event: "sync", + payload: { ...message, room: this.room, sender: this.sender }, + }); + } + + onMessage(cb: (msg: SyncMessage) => void): void { + this.listeners.add(cb); + } + + disconnect(): void { + this.listeners.clear(); + try { + this.client?.removeAllChannels(); + } catch { + /* ignore */ + } + } +} + +// --------------------------------------------------------------------------- // +// DashboardSync: binds a provider to a layout document. +// --------------------------------------------------------------------------- // +export class DashboardSync { + private provider: RealtimeProvider; + private room: string; + private widgets: Widget[] = []; + private onChange: (widgets: Widget[]) => void; + private lastRevision = 0; + private connected = false; + + constructor(opts: { + provider?: RealtimeProvider; + room?: string; + initial: Widget[]; + onChange: (widgets: Widget[]) => void; + }) { + this.provider = opts.provider ?? new BroadcastChannelProvider(); + this.room = opts.room ?? "default"; + this.widgets = opts.initial; + this.onChange = opts.onChange; + } + + async start(): Promise { + await this.provider.connect(this.room); + this.provider.onMessage((msg) => this._applyRemote(msg)); + // Announce our current state so late joiners converge. + this.provider.broadcast({ + kind: "init", + widgets: this.widgets, + revision: Date.now(), + }); + this.connected = true; + } + + private _applyRemote(msg: SyncMessage): void { + if (msg.revision < this.lastRevision) return; // ignore stale + this.lastRevision = msg.revision; + this.widgets = msg.widgets; + this.onChange(msg.widgets); + } + + /** Push a local layout change to every active session immediately. */ + update(widgets: Widget[]): void { + this.widgets = widgets; + this.lastRevision = Date.now(); + if (this.connected) { + this.provider.broadcast({ + kind: "update", + widgets, + revision: this.lastRevision, + }); + } + } + + get current(): Widget[] { + return this.widgets; + } + + stop(): void { + this.provider.disconnect(); + this.connected = false; + } +} + +/** Build the provider selected by `REALTIME_PROVIDER` (default: broadcast). */ +export function createProvider(kind = "broadcast"): RealtimeProvider { + if (kind === "yjs") return new YjsProvider(); + if (kind === "supabase") return new SupabaseProvider(); + return new BroadcastChannelProvider(); +} From cb8a4d606d79cf992615dad4b8378aa6c7fabab5 Mon Sep 17 00:00:00 2001 From: Raymond Oyondi Date: Sun, 12 Jul 2026 19:09:57 +0000 Subject: [PATCH 5/6] Document the four architectural improvements in README + .env.example --- .env.example | 22 ++++++++++++++++++++ README.md | 57 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 79 insertions(+) diff --git a/.env.example b/.env.example index 7e8638f..d45a352 100644 --- a/.env.example +++ b/.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= diff --git a/README.md b/README.md index 5e4c2e8..efefe4e 100644 --- a/README.md +++ b/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 From 5021ce64f8fa434c7a40df57456cf5595f9f3667 Mon Sep 17 00:00:00 2001 From: Raymond Oyondi Date: Sun, 12 Jul 2026 19:19:52 +0000 Subject: [PATCH 6/6] Fix type/serialization issues in realtime + edge-RAG modules - Shared in-memory fallback store keyed by cacheName (modelStreamCache) - Serialize int8 codes as plain arrays for JSON round-trip - Restore Int8Array on load of quantized index - Avoid parameter shadowing the worker quantize() fn - Make ONNXEmbedder.modelUrl/dim accessible; fix dynamic-import typing - Fix InMemoryLayoutRepository.save when creating a new layout --- backend/app/services/layout_repository.py | 2 +- src/lib/modelStreamCache.test.ts | 8 ++- src/lib/modelStreamCache.ts | 62 +++++++++++++---- src/lib/onnxWorker.ts | 42 +++++++----- src/lib/quantization.test.ts | 2 +- src/lib/quantization.ts | 5 +- src/lib/ragClient.ts | 83 ++++++++++++++++------- src/lib/realtimeSync.test.ts | 18 +++-- src/lib/realtimeSync.ts | 41 +++++++---- 9 files changed, 184 insertions(+), 79 deletions(-) diff --git a/backend/app/services/layout_repository.py b/backend/app/services/layout_repository.py index 735b636..a6b7647 100644 --- a/backend/app/services/layout_repository.py +++ b/backend/app/services/layout_repository.py @@ -67,7 +67,7 @@ def __init__(self) -> None: def save(self, layout: DashboardLayout) -> DashboardLayout: now = datetime.now(timezone.utc).isoformat() existing = self._store.get(layout.id) - layout.created_at = existing.created_at or now + layout.created_at = existing.created_at if existing else now layout.updated_at = now self._store[layout.id] = layout return layout diff --git a/src/lib/modelStreamCache.test.ts b/src/lib/modelStreamCache.test.ts index ed0de36..44281ec 100644 --- a/src/lib/modelStreamCache.test.ts +++ b/src/lib/modelStreamCache.test.ts @@ -40,7 +40,9 @@ describe("chunked, resumable model cache", () => { [1, 2, 3], [4, 5, 6], ]; - await cacheQuantizedIndex("semantic-index", index, { cacheName: "test-cache" }); + await cacheQuantizedIndex("semantic-index", index, { + cacheName: "test-cache", + }); const restored = await loadQuantizedIndex("semantic-index", { cacheName: "test-cache", }); @@ -57,7 +59,9 @@ describe("chunked, resumable model cache", () => { it("invalidates a model completely", async () => { const data = new Uint8Array(1024); - await cacheModel("to-del", data.buffer.slice(0), { cacheName: "test-cache" }); + await cacheModel("to-del", data.buffer.slice(0), { + cacheName: "test-cache", + }); await invalidateModel("to-del", { cacheName: "test-cache" }); expect(await loadModel("to-del", { cacheName: "test-cache" })).toBeNull(); }); diff --git a/src/lib/modelStreamCache.ts b/src/lib/modelStreamCache.ts index 7f49b7d..ad1636a 100644 --- a/src/lib/modelStreamCache.ts +++ b/src/lib/modelStreamCache.ts @@ -19,6 +19,13 @@ import { quantizeBatch, QuantizedVector } from "./quantization"; export const DEFAULT_CHUNK_SIZE = 256 * 1024; // 256 KiB +/** Serialized form of a quantized vector (codes as a plain array for JSON). */ +export interface SerializedQuantizedVector { + codes: number[]; + scale: number; + min: number; +} + export interface ModelManifest { name: string; totalChunks: number; @@ -26,7 +33,7 @@ export interface ModelManifest { chunkSize: number; etag: string; /** Optional quantized semantic index accompanying the model. */ - quantizedIndex?: QuantizedVector[]; + quantizedIndex?: SerializedQuantizedVector[]; } // --------------------------------------------------------------------------- // @@ -57,7 +64,7 @@ class CacheStorageStore implements ChunkStore { key, new Response(value, { headers: { "Content-Type": "application/octet-stream" }, - }) + }), ); } @@ -82,9 +89,25 @@ class CacheStorageStore implements ChunkStore { } } -/** In-memory fallback store (Node / SSR / tests). */ +/** + * In-memory fallback store (Node / SSR / tests). Stores are shared per + * `cacheName` via a module-level registry so that caching to a store and later + * loading from it (different `makeStore` calls) observe the same data — the + * same guarantee the Cache Storage API provides in the browser. + */ +const __memoryStores = new Map>(); + class MemoryStore implements ChunkStore { - private map = new Map(); + private map: Map; + + constructor(cacheName: string) { + let store = __memoryStores.get(cacheName); + if (!store) { + store = new Map(); + __memoryStores.set(cacheName, store); + } + this.map = store; + } async put(key: string, value: ArrayBuffer): Promise { this.map.set(key, value); @@ -109,7 +132,7 @@ function makeStore(cacheName: string): ChunkStore { } catch { /* fall through */ } - return new MemoryStore(); + return new MemoryStore(cacheName); } // --------------------------------------------------------------------------- // @@ -137,7 +160,7 @@ function simpleEtag(data: ArrayBuffer): string { export async function cacheModel( name: string, data: ArrayBuffer, - opts: { cacheName?: string; chunkSize?: number } = {} + opts: { cacheName?: string; chunkSize?: number } = {}, ): Promise { const chunkSize = opts.chunkSize ?? DEFAULT_CHUNK_SIZE; const store = makeStore(opts.cacheName ?? "gridify-models"); @@ -168,10 +191,16 @@ export async function cacheModel( export async function cacheQuantizedIndex( name: string, index: number[][], - opts: { cacheName?: string } = {} + opts: { cacheName?: string } = {}, ): Promise { const store = makeStore(opts.cacheName ?? "gridify-models"); - const quantized = quantizeBatch(index); + // Serialize codes as plain arrays so they survive the JSON round-trip + // (Int8Array loses its `.length` under JSON.stringify). + const quantized = quantizeBatch(index).map((q) => ({ + codes: Array.from(q.codes), + scale: q.scale, + min: q.min, + })); const manifest: ModelManifest = { name, totalChunks: 1, @@ -197,9 +226,9 @@ export async function loadModel( fetchChunk?: ( name: string, index: number, - manifest: ModelManifest + manifest: ModelManifest, ) => Promise; - } = {} + } = {}, ): Promise<{ data: ArrayBuffer; manifest: ModelManifest } | null> { const store = makeStore(opts.cacheName ?? "gridify-models"); const manifestBuf = await store.get(manifestKey(name)); @@ -226,19 +255,26 @@ export async function loadModel( /** Read a previously cached quantized index, if present. */ export async function loadQuantizedIndex( name: string, - opts: { cacheName?: string } = {} + opts: { cacheName?: string } = {}, ): Promise { const store = makeStore(opts.cacheName ?? "gridify-models"); const manifestBuf = await store.get(manifestKey(name)); if (!manifestBuf) return null; const manifest = decodeJSON(manifestBuf); - return manifest.quantizedIndex ?? null; + // Rebuild Int8Array codes from the plain arrays stored in the manifest. + const idx = manifest.quantizedIndex; + if (!idx) return null; + return idx.map((q) => ({ + codes: Int8Array.from(q.codes as unknown as number[]), + scale: q.scale, + min: q.min, + })); } /** Clear one model's chunks + manifest from the cache. */ export async function invalidateModel( name: string, - opts: { cacheName?: string } = {} + opts: { cacheName?: string } = {}, ): Promise { const store = makeStore(opts.cacheName ?? "gridify-models"); const manifestBuf = await store.get(manifestKey(name)); diff --git a/src/lib/onnxWorker.ts b/src/lib/onnxWorker.ts index 714ab17..6fd3c3f 100644 --- a/src/lib/onnxWorker.ts +++ b/src/lib/onnxWorker.ts @@ -15,9 +15,11 @@ async function init(modelUrl: string, modelDim = 64): Promise { } /** Quantize a float32 embedding to int8 to shrink the worker -> main payload. */ -function quantize( - vec: ArrayLike -): { codes: number[]; scale: number; min: number } { +function quantizeVec(vec: ArrayLike): { + codes: number[]; + scale: number; + min: number; +} { let min = Infinity; let max = -Infinity; for (let i = 0; i < vec.length; i++) { @@ -52,13 +54,16 @@ function hashEmbedding(text: string, modelDim: number): number[] { async function embed( texts: string[], - quantize = false -): Promise<{ embeddings: number[][]; quantized?: ReturnType[] }> { + returnQuantized = false, +): Promise<{ + embeddings: number[][]; + quantized?: ReturnType[]; +}> { if (!session) { throw new Error("ONNX worker not initialized"); } const results: number[][] = []; - const quantized: ReturnType[] = []; + const quantized: ReturnType[] = []; for (const text of texts) { const vec = hashEmbedding(text, dim); const tensor = { @@ -70,21 +75,26 @@ async function embed( const data = out[outputName].data as Float32Array; const full = Array.from(data); results.push(full); - if (quantize) quantized.push(quantize(full)); + if (returnQuantized) quantized.push(quantizeVec(full)); } - return { embeddings: results, quantized: quantize ? quantized : undefined }; + return { + embeddings: results, + quantized: returnQuantized ? quantized : undefined, + }; } export {}; -self.onmessage = async (e: MessageEvent<{ - type: string; - requestId: number; - modelUrl?: string; - modelDim?: number; - texts?: string[]; - quantize?: boolean; -}>) => { +self.onmessage = async ( + e: MessageEvent<{ + type: string; + requestId: number; + modelUrl?: string; + modelDim?: number; + texts?: string[]; + quantize?: boolean; + }>, +) => { const { type, requestId, modelUrl, modelDim, texts, quantize } = e.data; try { if (type === "init") { diff --git a/src/lib/quantization.test.ts b/src/lib/quantization.test.ts index 8d64434..f440fcd 100644 --- a/src/lib/quantization.test.ts +++ b/src/lib/quantization.test.ts @@ -46,7 +46,7 @@ describe("scalar quantization (float32 -> int8)", () => { it("reports a ~4x storage reduction", () => { const est = storageEstimate( - Array.from({ length: 100 }, () => new Array(64).fill(0)) + Array.from({ length: 100 }, () => new Array(64).fill(0)), ); expect(est.ratio).toBeCloseTo(4, 1); expect(est.int8Bytes * 4).toBe(est.float32Bytes); diff --git a/src/lib/quantization.ts b/src/lib/quantization.ts index 091b318..d175222 100644 --- a/src/lib/quantization.ts +++ b/src/lib/quantization.ts @@ -59,7 +59,10 @@ export function dequantizeBatch(vectors: QuantizedVector[]): number[][] { } /** Cosine similarity computed directly on int8 codes (avoids dequant round-trip). */ -export function quantizedCosine(a: QuantizedVector, b: QuantizedVector): number { +export function quantizedCosine( + a: QuantizedVector, + b: QuantizedVector, +): number { const n = Math.min(a.codes.length, b.codes.length); if (n === 0) return 0; let dot = 0; diff --git a/src/lib/ragClient.ts b/src/lib/ragClient.ts index db5ad94..a995a5e 100644 --- a/src/lib/ragClient.ts +++ b/src/lib/ragClient.ts @@ -16,7 +16,8 @@ * block the main UI thread. */ -import { cacheQuantizedIndex, dequantizeBatch, loadQuantizedIndex } from "./modelStreamCache"; +import { cacheQuantizedIndex, loadQuantizedIndex } from "./modelStreamCache"; +import { dequantizeBatch } from "./quantization"; export interface RAGChunk { id: string; @@ -53,7 +54,10 @@ export function cosineSimilarity(a: number[], b: number[]): number { * projected into `dim` buckets. Not semantically rich, but stable and good * enough for local pre-filtering before the cloud Chroma lookup. */ -export function hashEmbedding(text: string, dim: number = DEFAULT_DIM): number[] { +export function hashEmbedding( + text: string, + dim: number = DEFAULT_DIM, +): number[] { const vec = new Array(dim).fill(0); const tokens = text.toLowerCase().match(/[a-z0-9]+/g) ?? []; for (const tok of tokens) { @@ -78,12 +82,15 @@ export class ONNXEmbedder { private worker: Worker | null = null; private loading: Promise | null = null; private readonly modelUrl?: string; - private readonly dim: number; + readonly dim: number; private nextId = 0; - private pending = new Map void; - reject: (e: Error) => void; - }>(); + private pending = new Map< + number, + { + resolve: (v: number[][]) => void; + reject: (e: Error) => void; + } + >(); constructor(opts: { modelUrl?: string; dim?: number } = {}) { this.modelUrl = opts.modelUrl; @@ -109,21 +116,22 @@ export class ONNXEmbedder { private _loadWorker(): Promise { return new Promise((resolve, reject) => { try { - this.worker = new Worker( - new URL("./onnxWorker.ts", import.meta.url), - { type: "module" } - ); + this.worker = new Worker(new URL("./onnxWorker.ts", import.meta.url), { + type: "module", + }); } catch (err) { reject(err instanceof Error ? err : new Error(String(err))); return; } - this.worker.onmessage = (e: MessageEvent<{ - type: string; - requestId: number; - embeddings?: number[][]; - error?: string; - }>) => { + this.worker.onmessage = ( + e: MessageEvent<{ + type: string; + requestId: number; + embeddings?: number[][]; + error?: string; + }>, + ) => { const { type, requestId, embeddings, error } = e.data; const pending = this.pending.get(requestId); if (!pending) return; @@ -182,11 +190,26 @@ export class ONNXEmbedder { /** Local knowledge base used to build an index when the CDN cache is cold. */ export const LOCAL_DOCUMENTS: RAGChunk[] = [ - { id: "doc-temperature", text: "Temperature trends show normal cyclical variance with peak loads during afternoon operations." }, - { id: "doc-proxy", text: "Proxy devices maintain stable active flow control states with minor load offsets." }, - { id: "doc-node", text: "Marchival Arc node shows slightly higher average node temperature latency and should be investigated." }, - { id: "doc-uptime", text: "System uptime is steady at three hours ten minutes aggregate uptime with zero hard reboots." }, - { id: "doc-load", text: "Home Hub averages two point three thousand loads with optimal sensor load values." }, + { + id: "doc-temperature", + text: "Temperature trends show normal cyclical variance with peak loads during afternoon operations.", + }, + { + id: "doc-proxy", + text: "Proxy devices maintain stable active flow control states with minor load offsets.", + }, + { + id: "doc-node", + text: "Marchival Arc node shows slightly higher average node temperature latency and should be investigated.", + }, + { + id: "doc-uptime", + text: "System uptime is steady at three hours ten minutes aggregate uptime with zero hard reboots.", + }, + { + id: "doc-load", + text: "Home Hub averages two point three thousand loads with optimal sensor load values.", + }, ]; /** @@ -203,7 +226,9 @@ export class HybridRAG { /** Build an in-memory index from local documents (CDN cache is cold). */ private async _buildFromLocal(): Promise { - const embeddings = await this.embedder.embed(LOCAL_DOCUMENTS.map((d) => d.text)); + const embeddings = await this.embedder.embed( + LOCAL_DOCUMENTS.map((d) => d.text), + ); this.index = LOCAL_DOCUMENTS.map((d, i) => ({ ...d, embedding: embeddings[i], @@ -231,13 +256,17 @@ export class HybridRAG { if (remoteUrl) { try { - const res = await fetch(remoteUrl, { headers: { Accept: "application/json" } }); + const res = await fetch(remoteUrl, { + headers: { Accept: "application/json" }, + }); if (res.ok) { const payload = (await res.json()) as { chunks?: RAGChunk[] }; if (payload.chunks?.length) { const needEmbed = payload.chunks.filter((c) => !c.embedding); if (needEmbed.length) { - const emb = await this.embedder.embed(needEmbed.map((c) => c.text)); + const emb = await this.embedder.embed( + needEmbed.map((c) => c.text), + ); let i = 0; for (const c of payload.chunks) { if (!c.embedding) c.embedding = emb[i++]; @@ -259,7 +288,9 @@ export class HybridRAG { /** Quantize the current index to int8 and cache it for resumable reloads. */ private async _persistQuantizedIndex(): Promise { if (this.index.length === 0) return; - const vectors = this.index.map((c) => c.embedding ?? hashEmbedding(c.text, this.embedder.dim)); + const vectors = this.index.map( + (c) => c.embedding ?? hashEmbedding(c.text, this.embedder.dim), + ); try { await cacheQuantizedIndex("semantic-index", vectors); } catch { diff --git a/src/lib/realtimeSync.test.ts b/src/lib/realtimeSync.test.ts index ea3d33a..837f768 100644 --- a/src/lib/realtimeSync.test.ts +++ b/src/lib/realtimeSync.test.ts @@ -44,18 +44,22 @@ describe("BroadcastChannelProvider + DashboardSync", () => { local.stop(); }); - it("provider delivers both messages (staleness handled by DashboardSync)", async () => { + it("delivers messages between two providers in the same room", async () => { const received: SyncMessage[] = []; - const provider = new BroadcastChannelProvider(); - await provider.connect("room-2"); - provider.onMessage((m) => received.push(m)); + const a = new BroadcastChannelProvider(); + const b = new BroadcastChannelProvider(); + await a.connect("room-2"); + await b.connect("room-2"); + b.onMessage((m) => received.push(m)); - provider.broadcast({ kind: "update", widgets: widgetsB, revision: 100 }); - provider.broadcast({ kind: "update", widgets: widgetsA, revision: 50 }); + a.broadcast({ kind: "update", widgets: widgetsB, revision: 100 }); + a.broadcast({ kind: "update", widgets: widgetsA, revision: 50 }); await new Promise((r) => setTimeout(r, 0)); expect(received).toHaveLength(2); - provider.disconnect(); + expect(received.every((m) => m.sender === (a as any)["sender"])).toBe(true); + a.disconnect(); + b.disconnect(); }); it("createProvider returns the requested transport", () => { diff --git a/src/lib/realtimeSync.ts b/src/lib/realtimeSync.ts index a60d5ad..a80cd4c 100644 --- a/src/lib/realtimeSync.ts +++ b/src/lib/realtimeSync.ts @@ -53,7 +53,10 @@ function hubPublish(room: string, msg: SyncMessage): void { if (!subs) return; for (const cb of subs) cb(msg); } -function hubSubscribe(room: string, cb: (msg: SyncMessage) => void): () => void { +function hubSubscribe( + room: string, + cb: (msg: SyncMessage) => void, +): () => void { let subs = __hubs.get(room); if (!subs) { subs = new Set(); @@ -92,7 +95,11 @@ export class BroadcastChannelProvider implements RealtimeProvider { } broadcast(message: Omit): void { - const full: SyncMessage = { ...message, room: this.room, sender: this.sender }; + const full: SyncMessage = { + ...message, + room: this.room, + sender: this.sender, + }; if (this.channel) { this.channel.postMessage(full); } else { @@ -142,16 +149,21 @@ export class YjsProvider implements RealtimeProvider { let Y: any; let WebsocketProvider: any; try { - Y = await import("yjs"); - ({ WebsocketProvider } = await import("y-websocket")); + // Non-literal specifiers keep these optional deps out of the production + // bundle and let `tsc` pass when they are not installed. Install + // 'yjs' + 'y-websocket' to enable multiplayer CRDT sync. + const ySpec: string = "yjs"; + const wsSpec: string = "y-websocket"; + Y = await import(/* @vite-ignore */ ySpec); + ({ WebsocketProvider } = await import(/* @vite-ignore */ wsSpec)); } catch (err) { throw new Error( "YjsProvider requires 'yjs' and 'y-websocket' to be installed. " + - "Add them as dependencies to enable multiplayer CRDT sync." + "Add them as dependencies to enable multiplayer CRDT sync.", ); } this.doc = new Y.Doc(); - this.array = this.doc.getArray("widgets"); + this.array = (this.doc as any).getArray("widgets"); new WebsocketProvider(this.wsUrl, room, this.doc); this.array.observe(() => { const msg: SyncMessage = { @@ -205,16 +217,21 @@ export class SupabaseProvider implements RealtimeProvider { async connect(room: string): Promise { this.room = room; if (!this.url) return; // no-op when unconfigured - const mod = await import("@supabase/supabase-js").catch(() => null); + const supaSpec: string = "@supabase/supabase-js"; + const mod = await import(/* @vite-ignore */ supaSpec).catch(() => null); if (!mod) return; this.client = mod.createClient(this.url, this.anonKey); this.client .channel(`layout:${room}`) - .on("broadcast", { event: "sync" }, (payload: { payload: SyncMessage }) => { - if (payload.payload.sender !== this.sender) { - for (const cb of this.listeners) cb(payload.payload); - } - }) + .on( + "broadcast", + { event: "sync" }, + (payload: { payload: SyncMessage }) => { + if (payload.payload.sender !== this.sender) { + for (const cb of this.listeners) cb(payload.payload); + } + }, + ) .subscribe(); }