From 67bbe57e2607bbd7013de32ad4da4666ab8bb03f Mon Sep 17 00:00:00 2001
From: "Enot (ded) Skelly" <enot@mailbox.org>
Date: Wed, 27 May 2026 10:09:08 -0700
Subject: [PATCH] add brokers api endpoint

---
 cmd/tower/main.go                |  3 +--
 internal/api/handlers/brokers.go | 38 ++++++++++++++++++++++++++++++++
 internal/api/router/router.go    |  5 ++++-
 internal/ingest/ingest.go        | 26 ++++++++++++++++------
 4 files changed, 62 insertions(+), 10 deletions(-)
 create mode 100644 internal/api/handlers/brokers.go

diff --git a/cmd/tower/main.go b/cmd/tower/main.go
index 61135ea..ca681f6 100644
--- a/cmd/tower/main.go
+++ b/cmd/tower/main.go
@@ -124,9 +124,8 @@ func main() {
 
 	go broker1.Start(ctx)
 	go broker2.Start(ctx)
-
 	// ── HTTP server ──────────────────────────────────────────────────────────
-	r := router.New(h, store)
+	r := router.New(h, store, []*ingest.Worker{broker1, broker2})
 
 	srv := &http.Server{
 		Addr:    addr,
diff --git a/internal/api/handlers/brokers.go b/internal/api/handlers/brokers.go
new file mode 100644
index 0000000..a770399
--- /dev/null
+++ b/internal/api/handlers/brokers.go
@@ -0,0 +1,38 @@
+package handlers
+
+import (
+	"net/http"
+
+	"github.com/MeshCore-Tower/tower-server/internal/ingest"
+	"github.com/go-chi/chi/v5"
+)
+
+type BrokerStatus struct {
+	Name      string `json:"name"`
+	Connected bool   `json:"connected"`
+}
+
+// BrokersRouter mounts all /brokers routes onto a subrouter.
+//
+// GET  /brokers                              → ListBrokers
+//
+// Note: broker configuration is managed via the server config
+// file, not the API (v1). These endpoints are read-only.
+func BrokersRouter(workers []*ingest.Worker) http.Handler {
+	r := chi.NewRouter()
+
+	// GET  /brokers                              → ListBrokers
+	//
+	// Returns all configured brokers and their connection status
+	r.Get("/", func(w http.ResponseWriter, r *http.Request) {
+		brokers := make([]BrokerStatus, len(workers))
+		for i, v := range workers {
+			brokers[i] = BrokerStatus{
+				Name:      v.BrokerName(),
+				Connected: v.IsConnected(),
+			}
+		}
+		respond(w, http.StatusOK, brokers)
+	})
+	return r
+}
diff --git a/internal/api/router/router.go b/internal/api/router/router.go
index d796300..58213e9 100644
--- a/internal/api/router/router.go
+++ b/internal/api/router/router.go
@@ -10,6 +10,7 @@ import (
 	"github.com/MeshCore-Tower/tower-server/internal/api/handlers"
 	mw "github.com/MeshCore-Tower/tower-server/internal/api/middleware"
 	"github.com/MeshCore-Tower/tower-server/internal/hub"
+	"github.com/MeshCore-Tower/tower-server/internal/ingest"
 	"github.com/MeshCore-Tower/tower-server/internal/ws"
 )
 
@@ -21,6 +22,7 @@ import (
 //	/api/v1/           → public group
 //	  /packets         → packets subrouter
 //	  /nodes           → nodes subrouter
+//	  /brokers         → brokers subrouter
 //	  /observers       → observers subrouter
 //	  /channels        → channels subrouter
 //	  /iatas           → iatas subrouter
@@ -29,7 +31,7 @@ import (
 //
 // The private group is stubbed and ready for the auth middleware drop-in
 // described in Future Features → Admin authentication.
-func New(h *hub.Hub, reader api.Reader) http.Handler {
+func New(h *hub.Hub, reader api.Reader, workers []*ingest.Worker) http.Handler {
 	r := chi.NewRouter()
 
 	// ── Global middleware ────────────────────────────────────────────────────
@@ -49,6 +51,7 @@ func New(h *hub.Hub, reader api.Reader) http.Handler {
 		r.Group(func(r chi.Router) {
 			r.Mount("/packets", handlers.PacketsRouter())
 			r.Mount("/nodes", handlers.NodesRouter())
+			r.Mount("/brokers", handlers.BrokersRouter(workers))
 			r.Mount("/observers", handlers.ObserversRouter(reader))
 			r.Mount("/channels", handlers.ChannelsRouter(reader))
 			r.Mount("/messages", handlers.MessagesRouter(reader))
diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go
index 4f65816..42174f7 100644
--- a/internal/ingest/ingest.go
+++ b/internal/ingest/ingest.go
@@ -249,10 +249,11 @@ type ChannelKeyStore interface {
 
 // Worker holds the dependencies for one broker's ingest loop.
 type Worker struct {
-	cfg  Config
-	db   DB
-	hub  *hub.Hub
-	keys ChannelKeyStore
+	cfg    Config
+	db     DB
+	hub    *hub.Hub
+	keys   ChannelKeyStore
+	client mqtt.Client
 }
 
 // New creates an ingest Worker. Call Start() to connect and begin processing.
@@ -281,17 +282,28 @@ func (w *Worker) Start(ctx context.Context) {
 			log.Printf("ingest[%s]: connection lost: %v", w.cfg.BrokerName, err)
 		})
 
-	client := mqtt.NewClient(opts)
-	if tok := client.Connect(); tok.Wait() && tok.Error() != nil {
+	w.client = mqtt.NewClient(opts)
+	if tok := w.client.Connect(); tok.Wait() && tok.Error() != nil {
 		log.Printf("ingest[%s]: initial connect failed: %v", w.cfg.BrokerName, tok.Error())
 		// paho will retry; we fall through and wait for ctx
 	}
 
 	<-ctx.Done()
-	client.Disconnect(500)
+	w.client.Disconnect(500)
 	log.Printf("ingest[%s]: stopped", w.cfg.BrokerName)
 }
 
+func (w *Worker) BrokerName() string {
+	return w.cfg.BrokerName
+}
+
+func (w *Worker) IsConnected() bool {
+	if w.client == nil {
+		return false
+	}
+	return w.client.IsConnected()
+}
+
 // subscribe registers the wildcard topic handler after (re)connect.
 func (w *Worker) subscribe(client mqtt.Client) {
 	// meshcore/{IATA}/{pubkey}/packets