feat: compress dashboard shell over HTTP (#417)

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