Pulse/internal/api/connections_handlers.go

package api

import (
	"context"
	"encoding/json"
	"net/http"
	"strings"

	"github.com/rcourtman/pulse-go-rewrite/internal/config"
	"github.com/rcourtman/pulse-go-rewrite/internal/monitoring"
)

// ConnectionsHandlers serves the unified connections ledger. It does not own
// any persistence of its own — it composes per-type stores and the
// monitoring scheduler's in-memory health data into a single list.
type ConnectionsHandlers struct {
	getConfig                    func(ctx context.Context) *config.Config
	getPersistence               func(ctx context.Context) *config.ConfigPersistence
	getMonitor                   func(ctx context.Context) *monitoring.Monitor
	getTrueNASPoller             func(ctx context.Context) *monitoring.TrueNASPoller
	getVMwarePoller              func(ctx context.Context) *monitoring.VMwarePoller
	agentCommandSessionConnected func(organizationID, tokenID, agentID, hostname string) bool
}

// SetAgentCommandSessionProvider wires the live command-channel registry into
// the ledger. Telemetry liveness alone is not command readiness.
func (h *ConnectionsHandlers) SetAgentCommandSessionProvider(
	connected func(organizationID, tokenID, agentID, hostname string) bool,
) {
	if h == nil {
		return
	}
	h.agentCommandSessionConnected = connected
}

// NewConnectionsHandlers wires the aggregator behind the request-scoped
// tenant resolvers already used by ConfigHandlers, so the endpoint shares
// the same multi-tenant behavior as the per-type routes it aggregates.
func NewConnectionsHandlers(
	getConfig func(ctx context.Context) *config.Config,
	getPersistence func(ctx context.Context) *config.ConfigPersistence,
	getMonitor func(ctx context.Context) *monitoring.Monitor,
) *ConnectionsHandlers {
	return &ConnectionsHandlers{
		getConfig:      getConfig,
		getPersistence: getPersistence,
		getMonitor:     getMonitor,
	}
}

// SetPlatformPollers wires runtime poller summaries into the unified
// connections ledger for out-of-scheduler platform adapters.
func (h *ConnectionsHandlers) SetPlatformPollers(
	getTrueNASPoller func(ctx context.Context) *monitoring.TrueNASPoller,
	getVMwarePoller func(ctx context.Context) *monitoring.VMwarePoller,
) {
	if h == nil {
		return
	}
	h.getTrueNASPoller = getTrueNASPoller
	h.getVMwarePoller = getVMwarePoller
}

// HandleList returns every configured connection as a unified Connection row.
// No probing or network I/O happens here — state is derived purely from
// cached poller health.
func (h *ConnectionsHandlers) HandleList(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
		return
	}

	ctx := r.Context()
	cfg := h.getConfig(ctx)
	persistence := h.getPersistence(ctx)
	monitor := h.getMonitor(ctx)

	organizationID := resolveTenantOrgID(r)
	inputs := buildAggregatorInputsWithRuntimeSources(ctx, cfg, persistence, monitor, h.runtimeSources(ctx, organizationID))

	connections := buildConnections(inputs)
	if h.agentCommandSessionConnected != nil {
		for index := range connections {
			conn := &connections[index]
			if conn.Type != ConnectionTypeAgent || conn.AgentIdentity == nil {
				continue
			}
			connected := h.agentCommandSessionConnected(
				organizationID,
				conn.agentTokenID,
				conn.agentID,
				conn.AgentIdentity.Hostname,
			)
			conn.commandChannelConnected = &connected
			conn.Fleet.RemoteControl = connectionFleetRemoteControl(*conn)
			if conn.AgentIdentity.CommandsEnabled && !connected {
				desired := fleetStateUnknown
				if conn.Fleet.CommandPolicy != nil && strings.TrimSpace(conn.Fleet.CommandPolicy.Desired) != "" {
					desired = conn.Fleet.CommandPolicy.Desired
				}
				conn.Fleet.CommandPolicy = &ConnectionFleetCommandPolicy{
					Status:      fleetStateBlocked,
					Desired:     desired,
					Applied:     fleetStateEnabled,
					Enforcement: fleetStateBlocked,
					Reason:      "agent reports command execution enabled, but no admitted command channel is connected",
				}
			}
		}
	}
	writeJSON(w, http.StatusOK, ConnectionsListResponse{
		Connections: connections,
		Systems:     buildConnectionSystems(connections, monitor),
	})
}

func (h *ConnectionsHandlers) runtimeSources(ctx context.Context, orgID string) aggregatorRuntimeSources {
	runtime := aggregatorRuntimeSources{orgID: orgID}
	if h == nil {
		return runtime
	}
	if h.getTrueNASPoller != nil {
		runtime.truenasPoller = h.getTrueNASPoller(ctx)
	}
	if h.getVMwarePoller != nil {
		runtime.vmwarePoller = h.getVMwarePoller(ctx)
	}
	return runtime
}

// HandleProbe fingerprints a user-supplied address and returns the product
// types it detected, if any. No configuration is persisted — the caller
// uses the response to pick a credential slot in ConnectionEditor.
func (h *ConnectionsHandlers) HandleProbe(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
		return
	}

	r.Body = http.MaxBytesReader(w, r.Body, 4*1024)
	defer r.Body.Close()

	var req ProbeRequest
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeErrorResponse(w, http.StatusBadRequest, "invalid_request", "Invalid JSON body", nil)
		return
	}

	host, port, err := parseProbeAddress(req.Address)
	if err != nil {
		writeErrorResponse(w, http.StatusBadRequest, "invalid_address", err.Error(), nil)
		return
	}
	if err := validateProbeHost(r.Context(), host); err != nil {
		writeErrorResponse(w, http.StatusBadRequest, "invalid_address", err.Error(), nil)
		return
	}

	candidates, elapsed := runProbe(r.Context(), host, port, probeHTTPClient())
	writeJSON(w, http.StatusOK, ProbeResponse{
		Candidates: candidates,
		ProbedMs:   elapsed.Milliseconds(),
	})
}