HarborForge.Monitor/internal/bridge/bridge.go

// Package bridge provides a local HTTP server on MONITOR_PORT for
// communication between the HarborForge OpenClaw plugin and Monitor.
//
// The bridge serves two purposes:
//  1. Expose hardware telemetry to the plugin via GET /telemetry
//  2. Receive OpenClaw metadata from the plugin via POST /openclaw
//
// The bridge is optional: if monitorPort is 0 or not set, the bridge
// is not started and Monitor operates normally.
package bridge

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net"
	"net/http"
	"sync"
	"time"

	"git.hangman-lab.top/zhi/HarborForge.Monitor/internal/config"
	"git.hangman-lab.top/zhi/HarborForge.Monitor/internal/telemetry"
)

// OpenClawMeta holds metadata received from the OpenClaw plugin.
// This data is optional enrichment for heartbeat uploads.
type OpenClawMeta struct {
	Version       string `json:"version"`
	PluginVersion string `json:"plugin_version"`
	Agents        []any  `json:"agents,omitempty"`
}

// Server is the local bridge HTTP server.
type Server struct {
	cfg    config.Config
	logger *log.Logger
	srv    *http.Server

	mu          sync.RWMutex
	lastPayload *telemetry.Payload
	lastUpdated time.Time

	openclawMeta    *OpenClawMeta
	openclawUpdated time.Time
}

// New creates a bridge server. It does not start listening.
func New(cfg config.Config, logger *log.Logger) *Server {
	return &Server{
		cfg:    cfg,
		logger: logger,
	}
}

// UpdatePayload stores the latest telemetry payload so the bridge can
// serve it to plugin queries without re-collecting.
func (s *Server) UpdatePayload(p telemetry.Payload) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.lastPayload = &p
	s.lastUpdated = time.Now()
}

// bridgeResponse is the JSON structure served to the plugin.
type bridgeResponse struct {
	Status      string             `json:"status"`
	MonitorVer  string             `json:"monitor_version"`
	Identifier  string             `json:"identifier"`
	Telemetry   *telemetry.Payload `json:"telemetry,omitempty"`
	LastUpdated *time.Time         `json:"last_updated,omitempty"`
}

// GetOpenClawMeta returns the latest OpenClaw metadata received from
// the plugin, or nil if no metadata has been received.
func (s *Server) GetOpenClawMeta() *OpenClawMeta {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return s.openclawMeta
}

func (s *Server) handler() http.Handler {
	mux := http.NewServeMux()

	// Health / discovery endpoint
	mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/json")
		json.NewEncoder(w).Encode(map[string]string{
			"status":          "ok",
			"monitor_version": telemetry.Version,
			"identifier":      s.cfg.Identifier,
		})
	})

	// Telemetry endpoint — returns the latest cached payload
	mux.HandleFunc("/telemetry", func(w http.ResponseWriter, r *http.Request) {
		s.mu.RLock()
		payload := s.lastPayload
		updated := s.lastUpdated
		s.mu.RUnlock()

		resp := bridgeResponse{
			Status:     "ok",
			MonitorVer: telemetry.Version,
			Identifier: s.cfg.Identifier,
		}
		if payload != nil {
			resp.Telemetry = payload
			resp.LastUpdated = &updated
		}

		w.Header().Set("Content-Type", "application/json")
		json.NewEncoder(w).Encode(resp)
	})

	// OpenClaw metadata endpoint — plugin POSTs its metadata here
	mux.HandleFunc("/openclaw", func(w http.ResponseWriter, r *http.Request) {
		if r.Method != http.MethodPost {
			http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
			return
		}

		body, err := io.ReadAll(io.LimitReader(r.Body, 64*1024))
		if err != nil {
			http.Error(w, "read error", http.StatusBadRequest)
			return
		}
		defer r.Body.Close()

		var meta OpenClawMeta
		if err := json.Unmarshal(body, &meta); err != nil {
			http.Error(w, "invalid json", http.StatusBadRequest)
			return
		}

		s.mu.Lock()
		s.openclawMeta = &meta
		s.openclawUpdated = time.Now()
		s.mu.Unlock()

		s.logger.Printf("received OpenClaw metadata: version=%s plugin=%s agents=%d",
			meta.Version, meta.PluginVersion, len(meta.Agents))

		w.Header().Set("Content-Type", "application/json")
		json.NewEncoder(w).Encode(map[string]string{
			"status": "ok",
		})
	})

	return mux
}

// Start begins listening on 127.0.0.1:<monitorPort>. It blocks until
// the context is cancelled or an error occurs.
func (s *Server) Start(ctx context.Context) error {
	if s.cfg.MonitorPort <= 0 {
		return nil // bridge disabled
	}

	addr := fmt.Sprintf("127.0.0.1:%d", s.cfg.MonitorPort)
	listener, err := net.Listen("tcp", addr)
	if err != nil {
		return fmt.Errorf("bridge listen on %s: %w", addr, err)
	}

	s.srv = &http.Server{
		Handler:      s.handler(),
		ReadTimeout:  5 * time.Second,
		WriteTimeout: 5 * time.Second,
		IdleTimeout:  30 * time.Second,
	}

	s.logger.Printf("bridge listening on %s", addr)

	go func() {
		<-ctx.Done()
		shutCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
		defer cancel()
		s.srv.Shutdown(shutCtx)
	}()

	if err := s.srv.Serve(listener); err != nil && err != http.ErrServerClosed {
		return fmt.Errorf("bridge serve: %w", err)
	}
	return nil
}