From 4c2b94a5f0b22024ec2a9cc42c4e079a0f56ee4b Mon Sep 17 00:00:00 2001
From: Hanzo AI <dev@hanzo.ai>
Date: Thu, 9 Apr 2026 17:46:06 -0700
Subject: [PATCH] Move admin HTTP routes to /v1/stream/ prefix, add /healthz

Admin endpoints now served under /v1/stream/{status,topics,groups}
following the /<version>/<service>/<path> convention. Root /
redirects to /v1/stream/. Added /healthz for K8s HTTP probes.
---
 LLM.md            | 13 +++++++++++++
 protocol/admin.go | 39 ++++++++++++++++++++++++++++++---------
 2 files changed, 43 insertions(+), 9 deletions(-)

diff --git a/LLM.md b/LLM.md
index 11f6f92..8f10c34 100644
--- a/LLM.md
+++ b/LLM.md
@@ -83,11 +83,24 @@ kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test
 kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
 ```
 
+## Admin HTTP API (port 9093)
+
+All management endpoints under `/v1/stream/`:
+```
+GET /v1/stream/status  — service status (JSON)
+GET /v1/stream/topics  — list topics with partition details
+GET /v1/stream/groups  — list consumer group offsets
+GET /healthz           — health check (K8s probes)
+```
+
+Root `/` redirects to `/v1/stream/`.
+
 ## Deployment (hanzo-k8s, do-sfo3)
 ```
 Namespace: hanzo
 PubSub:    pubsub.hanzo.svc:4222   (nats:2.10-alpine, 1 replica, 20Gi PVC)
 Stream:    stream.hanzo.svc:9092   (ghcr.io/hanzoai/stream:latest, 2 replicas)
+Admin:     :9093 (HTTP, /healthz for K8s probes)
 ```
 Dockerfile builds linux/amd64 via `GOARCH=amd64`. CI pushes to GHCR on every main push.
 
diff --git a/protocol/admin.go b/protocol/admin.go
index 49b0c38..6f41241 100644
--- a/protocol/admin.go
+++ b/protocol/admin.go
@@ -46,17 +46,23 @@ type GroupOffset struct {
 	Offset    int64  `json:"offset"`
 }
 
-// StartAdmin starts the admin HTTP server on the configured port
+// StartAdmin starts the admin HTTP server on the configured port.
+// All management endpoints are served under /v1/stream/.
+// /healthz is served at root for K8s probes.
 func (b *Broker) StartAdmin() {
 	if b.Config.AdminPort == 0 {
 		return
 	}
 
+	const prefix = "/v1/stream"
+
 	mux := http.NewServeMux()
-	mux.HandleFunc("/status", b.handleStatus)
-	mux.HandleFunc("/topics", b.handleTopics)
-	mux.HandleFunc("/groups", b.handleGroups)
-	mux.HandleFunc("/", b.handleIndex)
+	mux.HandleFunc(prefix+"/status", b.handleStatus)
+	mux.HandleFunc(prefix+"/topics", b.handleTopics)
+	mux.HandleFunc(prefix+"/groups", b.handleGroups)
+	mux.HandleFunc("/healthz", b.handleHealthz)
+	mux.HandleFunc(prefix+"/", b.handleIndex)
+	mux.HandleFunc("/", b.handleRoot)
 
 	addr := fmt.Sprintf(":%d", b.Config.AdminPort)
 	log.Info("Admin HTTP server listening on %s", addr)
@@ -67,16 +73,31 @@ func (b *Broker) StartAdmin() {
 	}()
 }
 
-func (b *Broker) handleIndex(w http.ResponseWriter, r *http.Request) {
+func (b *Broker) handleRoot(w http.ResponseWriter, r *http.Request) {
 	if r.URL.Path != "/" {
 		http.NotFound(w, r)
 		return
 	}
+	http.Redirect(w, r, "/v1/stream/", http.StatusMovedPermanently)
+}
+
+func (b *Broker) handleIndex(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Content-Type", "text/plain")
 	fmt.Fprintf(w, "Hanzo Stream Admin\n\n")
-	fmt.Fprintf(w, "GET /status  — service health\n")
-	fmt.Fprintf(w, "GET /topics  — list topics with partition details\n")
-	fmt.Fprintf(w, "GET /groups  — list consumer group offsets\n")
+	fmt.Fprintf(w, "GET /v1/stream/status  — service status\n")
+	fmt.Fprintf(w, "GET /v1/stream/topics  — list topics with partition details\n")
+	fmt.Fprintf(w, "GET /v1/stream/groups  — list consumer group offsets\n")
+	fmt.Fprintf(w, "GET /healthz           — health check\n")
+}
+
+func (b *Broker) handleHealthz(w http.ResponseWriter, r *http.Request) {
+	if b.PubSub != nil && b.PubSub.NC != nil && b.PubSub.NC.IsConnected() {
+		w.WriteHeader(http.StatusOK)
+		fmt.Fprintf(w, "ok\n")
+		return
+	}
+	w.WriteHeader(http.StatusServiceUnavailable)
+	fmt.Fprintf(w, "pubsub disconnected\n")
 }
 
 func (b *Broker) handleStatus(w http.ResponseWriter, r *http.Request) {