src/compression.py

@@ -0,0 +1,45 @@
+from starlette.datastructures import Headers
+from starlette.middleware.gzip import GZipResponder, IdentityResponder
+from starlette.types import ASGIApp, Receive, Scope, Send
+
+
+def _quality(accept_encoding: str, coding: str) -> float:
+    qualities: dict[str, float] = {}
+    for entry in accept_encoding.split(","):
+        parts = [part.strip() for part in entry.split(";")]
+        name = parts[0].lower()
+        if not name:
+            continue
+        quality = 1.0
+        for parameter in parts[1:]:
+            key, separator, value = parameter.partition("=")
+            if separator and key.strip().lower() == "q":
+                try:
+                    quality = float(value.strip())
+                except ValueError:
+                    quality = 0.0
+        qualities[name] = quality
+    return qualities.get(coding, qualities.get("*", 0.0))
+
+
+class NegotiatedGZipMiddleware:
+    """Compress substantial responses only when the client permits gzip."""
+
+    def __init__(self, app: ASGIApp, minimum_size: int = 500, compresslevel: int = 9):
+        self.app = app
+        self.minimum_size = minimum_size
+        self.compresslevel = compresslevel
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+
+        accepted = Headers(scope=scope).get("Accept-Encoding", "")
+        if _quality(accepted, "gzip") > 0:
+            responder: ASGIApp = GZipResponder(
+                self.app, self.minimum_size, compresslevel=self.compresslevel
+            )
+        else:
+            responder = IdentityResponder(self.app, self.minimum_size)
+        await responder(scope, receive, send)

src/main.py

@@ -19,6 +19,7 @@ from fastapi.staticfiles import StaticFiles
 from pydantic import BaseModel, Field, PositiveInt, field_validator, model_validator
 
 from src import dashboard_auth, gitea_proxy
+from src.compression import NegotiatedGZipMiddleware
 from src.gitea_proxy import (
     activity_events,
     current_user,
@@ -67,6 +68,7 @@ async def lifespan(_app: FastAPI):
 
 app = FastAPI(title="Stackchain Dashboard", lifespan=lifespan)
 app.add_middleware(RequestBodyLimitMiddleware, limit_for=request_body_limit)
+app.add_middleware(NegotiatedGZipMiddleware, minimum_size=1_024)
 CONTEXT_TIMEOUT_SECONDS = 5.0
 EVENT_STREAM_TIMEOUT_SECONDS = 5.0
 READINESS_TIMEOUT_SECONDS = 5.0

tests/test_frontend_bundle.py

@@ -3,8 +3,10 @@ import re
 import shutil
 from pathlib import Path
 
+import httpx
 import pytest
 
+from src import main
 from src.frontend_bundle import build_frontend
 from src.views import dashboard, runtime_bundle, service_worker
 
@@ -37,6 +39,62 @@ def test_page_runtime_is_one_deterministic_content_addressed_bundle(tmp_path):
     assert changed.runtime_bytes != first.runtime_bytes
 
 
+@pytest.mark.anyio
+async def test_runtime_is_transferred_as_gzip_when_the_client_accepts_it():
+    transport = httpx.ASGITransport(app=main.app)
+
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        response = await client.get(
+            f"/{main.FRONTEND_BUILD.runtime_name}",
+            headers={"Accept-Encoding": "gzip"},
+        )
+
+    assert response.status_code == 200
+    assert response.headers["content-encoding"] == "gzip"
+    assert response.headers["vary"] == "Accept-Encoding"
+    assert int(response.headers["content-length"]) <= 100 * 1024
+    assert response.content == main.FRONTEND_BUILD.runtime_bytes
+    assert response.headers["cache-control"] == "public, max-age=31536000, immutable"
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("accept_encoding", ["identity", "gzip;q=0, identity;q=1"])
+async def test_runtime_stays_unencoded_when_the_client_declines_gzip(accept_encoding):
+    transport = httpx.ASGITransport(app=main.app)
+
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        response = await client.get(
+            f"/{main.FRONTEND_BUILD.runtime_name}",
+            headers={"Accept-Encoding": accept_encoding},
+        )
+
+    assert response.status_code == 200
+    assert "content-encoding" not in response.headers
+    assert int(response.headers["content-length"]) == len(main.FRONTEND_BUILD.runtime_bytes)
+    assert response.content == main.FRONTEND_BUILD.runtime_bytes
+
+
+@pytest.mark.anyio
+async def test_dashboard_shell_is_compressed_without_losing_response_policies():
+    transport = httpx.ASGITransport(app=main.app)
+
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        page = await client.get("/", headers={"Accept-Encoding": "gzip"})
+        stylesheet = await client.get(
+            "/static/dashboard.css", headers={"Accept-Encoding": "gzip"}
+        )
+        health = await client.get("/healthz", headers={"Accept-Encoding": "gzip"})
+
+    assert page.headers["content-encoding"] == "gzip"
+    assert page.headers["cache-control"] == "no-cache"
+    assert page.headers["content-security-policy"]
+    assert main.FRONTEND_BUILD.runtime_name in page.text
+    assert stylesheet.headers["content-encoding"] == "gzip"
+    assert "mobile-task-dock" in stylesheet.text
+    assert health.json() == {"service": "stackchain-dashboard", "status": "ok"}
+    assert "content-encoding" not in health.headers
+
+
 def test_offline_shell_precaches_exact_runtime_without_superseded_page_modules():
     build = build_frontend(FRONTEND)