feat: ADD health endpoint with simple testing #29

app.py

@@ -1,6 +1,6 @@
 # aukpad.py
 from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request, HTTPException
-from fastapi.responses import HTMLResponse, RedirectResponse, PlainTextResponse, FileResponse
+from fastapi.responses import HTMLResponse, RedirectResponse, PlainTextResponse, FileResponse, JSONResponse
 import json, re, secrets, string, time, os, threading, asyncio, hashlib
 from collections import defaultdict
 from typing import Optional
@@ -17,6 +17,7 @@ RETENTION_HOURS = int(os.getenv("RETENTION_HOURS", "48"))  # Default 48 hours
 MAX_ROOMS = int(os.getenv("MAX_ROOMS", "10000"))
 TRUST_PROXY = os.getenv("TRUST_PROXY", "false").lower() == "true"
 DESCRIPTION = os.getenv("DESCRIPTION", "powered by aukpad.com")
+START_TIME = time.time()
 
 DOC_ID_RE = re.compile(r"^[a-zA-Z0-9_-]{1,64}$")
 
@@ -540,6 +541,41 @@ ta.addEventListener("input", () => {
 def favicon():
     return FileResponse("favicon.ico")
 
+@app.get("/health", include_in_schema=False)
+def health():
+    checks = {"app": "ok"}
+    healthy = True
+
+    # Writable = can we still take new pads, or are we at room capacity?
+    checks["write"] = "ok" if len(rooms) < MAX_ROOMS else "at capacity"
+    healthy = checks["write"] == "ok"
+
+    # Cache: real round-trip (write, read back, delete) so a half-dead
+    # connection shows up instead of just "configured".
+    if not USE_VALKEY:
+        checks["cache"] = "disabled"
+    elif redis_client is None:
+        checks["cache"] = "unavailable"
+        healthy = False
+    else:
+        key = f"health:{random_id()}"
+        try:
+            redis_client.setex(key, 10, "ok")
+            checks["cache"] = "ok" if redis_client.get(key) == "ok" else "readback failed"
+            redis_client.delete(key)
+        except Exception as e:
+            # Type only — the exception text can contain the connection URL.
+            checks["cache"] = f"error: {type(e).__name__}"
+        if checks["cache"] != "ok":
+            healthy = False
+
+    return JSONResponse({
+        "status": "ok" if healthy else "degraded",
+        "checks": checks,
+        "rooms": len(rooms),
+        "uptime_seconds": round(time.time() - START_TIME),
+    }, status_code=200 if healthy else 503)
+
 @app.get("/system/info", response_class=HTMLResponse)
 def get_system_info():
     max_text_size_mb = int(os.getenv("MAX_TEXT_SIZE", "5"))