feat(ws): make resolved paths a config toggle in session

README.md

@@ -268,6 +268,23 @@ everything). Empty array means match nothing on that dimension. `regionIds` and
 }
 ```
 
+**Configure** — toggle connection-wide settings. Currently just `resolvePath`,
+which enables per-hop node resolution on `packetObservation` events (see
+below). Unlike `subscribe`, this is a single flag for the whole connection,
+not additive across calls — each `configure` sets it to exactly the value
+sent, and it can be flipped on or off as many times as you like for the life
+of the connection. Default is `false`.
+
+```json
+{ "v": 1, "type": "configure", "id": "cfg-1", "resolvePath": true }
+```
+
+The server replies:
+
+```json
+{ "v": 1, "type": "configured", "id": "cfg-1", "resolvePath": true }
+```
+
 **Ping**
 
 ```json
@@ -283,6 +300,14 @@ everything). Empty array means match nothing on that dimension. `regionIds` and
 | `nodeUpdate`        | Node upserted from advert                           |
 | `channelMessage`    | Decrypted channel message (scope must include hash) |
 
+When `resolvePath` is enabled via `configure`, `packetObservation` events
+include a `resolvedPath` array: one entry per hop in the packet's path, each
+with a `confidence` (`"high"` for exactly one candidate node, `"ambiguous"`
+for multiple, `"none"` for zero) and the matching node(s)' id, name,
+lat/lng, and public key. Without `resolvePath` enabled, `resolvedPath` is
+`null` — this keeps the default event payload small; only opt in if you're
+actually rendering path connections (e.g. drawing hops on a map).
+
 ### Backpressure
 
 The server write buffer per connection is bounded at 256 events. If a client

internal/hub/hub.go

@@ -34,9 +34,15 @@ const (
 
 // Event is a single fan-out unit. Payload is pre-serialised JSON so the
 // broadcast loop never touches encoding — it's done once by the ingest path.
+//
+// PayloadResolved is an optional second serialization carrying additional
+// fields for clients that opted into them via configure (currently just
+// resolvedPath on packetObservation events). Left nil for event types that
+// don't have an opt-in variant; the hub falls back to Payload in that case.
 type Event struct {
-	Type    EventType
-	Payload json.RawMessage
+	Type            EventType
+	Payload         json.RawMessage
+	PayloadResolved json.RawMessage
 
 	// Routing metadata used by the hub to match subscriptions.
 	// Populated by the ingest layer before calling Broadcast.
@@ -64,6 +70,13 @@ type Client struct {
 	Send          chan Event
 	laggedCH      chan LaggedNotification
 	subscriptions map[string]Scope // OR semantics: event matches if it matches any scope entry
+
+	// ResolvePath is a connection-wide opt-in (not per-subscription), set via
+	// SetResolvePath ("configure" WS messages). Freely toggleable at any
+	// point during the connection's lifetime. Only ever read/written inside
+	// Run(), so it needs no locking despite Client being shared with the WS
+	// goroutines.
+	ResolvePath bool
 }
 
 // matches returns true if the event satisfies at least one of the client's
@@ -106,6 +119,7 @@ func scopeMatches(s Scope, e Event) bool {
 type Hub struct {
 	subscribe   chan subscribeMsg
 	unsubscribe chan unsubscribeMsg
+	configure   chan configureMsg
 	remove      chan *Client
 	broadcast   chan Event
 }
@@ -121,11 +135,20 @@ type unsubscribeMsg struct {
 	subscriptionID string
 }
 
+// configureMsg carries a connection-wide setting change, decoupled from the
+// subscribe/unsubscribe scope mechanics so it can be toggled independently
+// and repeatedly over the life of a connection.
+type configureMsg struct {
+	client      *Client
+	resolvePath bool
+}
+
 // New creates a Hub. Call Run() in a goroutine before using it.
 func New() *Hub {
 	return &Hub{
 		subscribe:   make(chan subscribeMsg, 64),
 		unsubscribe: make(chan unsubscribeMsg, 64),
+		configure:   make(chan configureMsg, 64),
 		remove:      make(chan *Client, 64),
 		broadcast:   make(chan Event, 512),
 	}
@@ -158,6 +181,15 @@ func (h *Hub) RemoveScope(c *Client, id string) {
 	h.unsubscribe <- unsubscribeMsg{client: c, subscriptionID: id}
 }
 
+// SetResolvePath toggles a client's opt-in to the resolvedPath variant of
+// packetObservation events. Unlike scopes, this is a single connection-wide
+// flag (not additive/OR'd) and can be flipped on or off at any point during
+// the connection's lifetime — takes effect on the next broadcast after the
+// hub processes it.
+func (h *Hub) SetResolvePath(c *Client, enabled bool) {
+	h.configure <- configureMsg{client: c, resolvePath: enabled}
+}
+
 // Remove deregisters a client and closes its Send channel.
 // Safe to call from any goroutine (e.g. the WS handler's defer).
 func (h *Hub) Remove(c *Client) {
@@ -203,6 +235,11 @@ func (h *Hub) Run() {
 				delete(msg.client.subscriptions, msg.subscriptionID)
 			}
 
+		case msg := <-h.configure:
+			if _, ok := clients[msg.client]; ok {
+				msg.client.ResolvePath = msg.resolvePath
+			}
+
 		case c := <-h.remove:
 			if _, ok := clients[c]; ok {
 				delete(clients, c)
@@ -215,8 +252,12 @@ func (h *Hub) Run() {
 				if !c.matches(evt) {
 					continue
 				}
+				outEvt := evt
+				if c.ResolvePath && evt.PayloadResolved != nil {
+					outEvt.Payload = evt.PayloadResolved
+				}
 				select {
-				case c.Send <- evt:
+				case c.Send <- outEvt:
 				default:
 					dropped := 1
 					select {

internal/hub/hub_test.go

@@ -4,6 +4,7 @@
 package hub
 
 import (
+	"encoding/json"
 	"testing"
 	"time"
 )
@@ -245,3 +246,118 @@ func TestClientMatches_ORSemantics(t *testing.T) {
 		t.Error("expected no match for unsubscribed event type")
 	}
 }
+
+func TestHub_ResolvePath_OptedIn_GetsResolvedPayload(t *testing.T) {
+	h := runHub(t)
+	c := h.NewClient()
+	h.AddScope(c, "sub1", Scope{Events: []EventType{EventPacketObservation}})
+	h.SetResolvePath(c, true)
+
+	time.Sleep(10 * time.Millisecond)
+
+	h.Broadcast(Event{
+		Type:            EventPacketObservation,
+		IATA:            "YVR",
+		Payload:         json.RawMessage(`{"resolvedPath":null}`),
+		PayloadResolved: json.RawMessage(`{"resolvedPath":[{"confidence":"high"}]}`),
+	})
+
+	select {
+	case evt := <-c.Send:
+		if string(evt.Payload) != `{"resolvedPath":[{"confidence":"high"}]}` {
+			t.Errorf("expected resolved payload, got %s", evt.Payload)
+		}
+	case <-time.After(100 * time.Millisecond):
+		t.Fatal("expected event, timed out")
+	}
+}
+
+func TestHub_ResolvePath_DefaultOff_GetsBasePayload(t *testing.T) {
+	h := runHub(t)
+	c := h.NewClient()
+	h.AddScope(c, "sub1", Scope{Events: []EventType{EventPacketObservation}})
+	// no SetResolvePath call — default is off
+
+	time.Sleep(10 * time.Millisecond)
+
+	h.Broadcast(Event{
+		Type:            EventPacketObservation,
+		IATA:            "YVR",
+		Payload:         json.RawMessage(`{"resolvedPath":null}`),
+		PayloadResolved: json.RawMessage(`{"resolvedPath":[{"confidence":"high"}]}`),
+	})
+
+	select {
+	case evt := <-c.Send:
+		if string(evt.Payload) != `{"resolvedPath":null}` {
+			t.Errorf("expected base payload (not opted in), got %s", evt.Payload)
+		}
+	case <-time.After(100 * time.Millisecond):
+		t.Fatal("expected event, timed out")
+	}
+}
+
+func TestHub_ResolvePath_OptedIn_NoResolvedVariant_FallsBackToBase(t *testing.T) {
+	h := runHub(t)
+	c := h.NewClient()
+	// e.g. nodeUpdate events never carry a PayloadResolved variant
+	h.AddScope(c, "sub1", Scope{Events: []EventType{EventNodeUpdate}})
+	h.SetResolvePath(c, true)
+
+	time.Sleep(10 * time.Millisecond)
+
+	h.Broadcast(Event{
+		Type:    EventNodeUpdate,
+		IATA:    "YVR",
+		Payload: json.RawMessage(`{"nodeId":"abc"}`),
+		// PayloadResolved intentionally left nil
+	})
+
+	select {
+	case evt := <-c.Send:
+		if string(evt.Payload) != `{"nodeId":"abc"}` {
+			t.Errorf("expected base payload as fallback, got %s", evt.Payload)
+		}
+	case <-time.After(100 * time.Millisecond):
+		t.Fatal("expected event, timed out")
+	}
+}
+
+func TestHub_ResolvePath_ToggleableLive(t *testing.T) {
+	h := runHub(t)
+	c := h.NewClient()
+	h.AddScope(c, "sub1", Scope{Events: []EventType{EventPacketObservation}})
+
+	broadcastAndRead := func() string {
+		h.Broadcast(Event{
+			Type:            EventPacketObservation,
+			IATA:            "YVR",
+			Payload:         json.RawMessage(`{"resolvedPath":null}`),
+			PayloadResolved: json.RawMessage(`{"resolvedPath":[{"confidence":"high"}]}`),
+		})
+		select {
+		case evt := <-c.Send:
+			return string(evt.Payload)
+		case <-time.After(100 * time.Millisecond):
+			t.Fatal("expected event, timed out")
+			return ""
+		}
+	}
+
+	time.Sleep(10 * time.Millisecond)
+	if got := broadcastAndRead(); got != `{"resolvedPath":null}` {
+		t.Errorf("expected base payload before opting in, got %s", got)
+	}
+
+	h.SetResolvePath(c, true)
+	time.Sleep(10 * time.Millisecond)
+	if got := broadcastAndRead(); got != `{"resolvedPath":[{"confidence":"high"}]}` {
+		t.Errorf("expected resolved payload after opting in, got %s", got)
+	}
+
+	h.SetResolvePath(c, false)
+	time.Sleep(10 * time.Millisecond)
+	if got := broadcastAndRead(); got != `{"resolvedPath":null}` {
+		t.Errorf("expected base payload after opting back out, got %s", got)
+	}
+}

internal/ingest/ingest.go

@@ -309,6 +309,34 @@ func (w *Worker) broadcast(eventType hub.EventType, iata string, payloadType uin
 	})
 }
 
+// broadcastPacketObservation marshals evt twice: once as-is (the default
+// payload every packetObservation subscriber gets) and once with
+// resolvedPath populated (delivered only to connections that opted in via
+// the "configure" WS message; see hub.Client.ResolvePath). resolvedPath is
+// passed in rather than computed here because the caller already has the
+// path-hash resolution results in hand from other per-packet work (known
+// route detection, capability detection) — this adds no extra DB calls.
+func (w *Worker) broadcastPacketObservation(iata string, payloadType uint8, evt packetObservationEvent, resolvedPath []api.ResolvedHop) {
+	base, err := json.Marshal(evt)
+	if err != nil {
+		log.Printf("ingest[%s]: failed to marshal packetObservation event: %v", w.cfg.BrokerName, err)
+		return
+	}
+	evt.Observation.ResolvedPath = resolvedPath
+	resolved, err := json.Marshal(evt)
+	if err != nil {
+		log.Printf("ingest[%s]: failed to marshal packetObservation event (resolved variant): %v", w.cfg.BrokerName, err)
+		resolved = nil // fall back to base-only; not fatal
+	}
+	w.hub.Broadcast(hub.Event{
+		Type:            hub.EventPacketObservation,
+		Payload:         base,
+		PayloadResolved: resolved,
+		IATA:            iata,
+		PayloadType:     payloadType,
+	})
+}
+
 // parseNumber handles RSSI and SNR fields that different observer types send as
 // either a bare JSON number (e.g. -108) or a quoted string (e.g. "-108").
 // Returns 0 if the value is missing or unparseable.

internal/ingest/packet.go

@@ -16,7 +16,6 @@ import (
 	"time"
 
 	"github.com/MeshCore-Beacon/beacon-server/internal/api"
-	"github.com/MeshCore-Beacon/beacon-server/internal/hub"
 	"github.com/google/uuid"
 	"github.com/meshcore-go/meshcore-go"
 )
@@ -94,7 +93,7 @@ type packetObservationEvent struct {
 			HopCount uint8  `json:"hopCount"`
 		} `json:"pathLength"`
 		PropagationTimeMs int32             `json:"propagationTimeMs"`
-		ResolvedPath      []api.ResolvedHop `json:"resolvedPath"`
+		ResolvedPath      []api.ResolvedHop `json:"resolvedPath"` // only present in the resolvePath-opted-in variant; see hub.Event.PayloadResolved
 	} `json:"observation"`
 }
 
@@ -815,7 +814,6 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [
 		evt.Observation.PathLength.HashSize = packet.PathHashSize()
 		evt.Observation.PathLength.HopCount = packet.PathHashCount()
 		evt.Observation.PropagationTimeMs = 0 // not yet calculated
-		evt.Observation.ResolvedPath = api.BuildResolvedPath(hashes, resolved)
 		count, err := w.db.GetPacketObservationCount(ctx, packetHash[:])
 		if err != nil {
 			log.Printf("ingest[%s]: failed to get observation count: %v", w.cfg.BrokerName, err)
@@ -825,7 +823,8 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [
 		if matchedScope != nil {
 			evt.Packet.Scope = matchedScope
 		}
-		w.broadcast(hub.EventPacketObservation, iata, packet.PayloadType(), "", evt)
+		resolvedPath := api.BuildResolvedPath(hashes, resolved)
+		w.broadcastPacketObservation(iata, packet.PayloadType(), evt, resolvedPath)
 	}
 }
 

internal/ws/handler.go

@@ -8,9 +8,16 @@
 //	On connect: server sends hello { v:1, type:"hello", serverTime:<ms>, connectionId:"uuid" }
 //
 //	Client → Server:
-//	  subscribe   { v, type, id, scope }   → server replies subscribed { v, type, id, subscriptionId }
+//	  subscribe   { v, type, id, scope }         → server replies subscribed { v, type, id, subscriptionId }
 //	  unsubscribe { v, type, id, subscriptionId }
-//	  ping        { v, type, id }          → server replies pong { v, type, id }
+//	  configure   { v, type, id, resolvePath }   → server replies configured { v, type, id, resolvePath }
+//	  ping        { v, type, id }                → server replies pong { v, type, id }
+//
+//	configure's resolvePath (bool) is a connection-wide setting, not
+//	per-subscription: enables/disables per-hop resolvedPath data on
+//	packetObservation events. Freely toggleable at any point during the
+//	connection; each configure call sets it to exactly the value sent
+//	(not additive across calls, unlike subscribe scopes). Default false.
 //
 //	Server → Client events (unsolicited):
 //	  packetObservation, observerStatus, nodeUpdate, channelMessage
@@ -152,6 +159,11 @@ type clientMessage struct {
 	ID             string          `json:"id"`
 	SubscriptionID string          `json:"subscriptionId,omitempty"`
 	Scope          *subscribeScope `json:"scope,omitempty"`
+
+	// ResolvePath is only read for "configure" messages: enables/disables
+	// the resolvedPath variant of packetObservation events for the whole
+	// connection. See package doc.
+	ResolvePath bool `json:"resolvePath,omitempty"`
 }
 
 // subscribeScope mirrors the scope object in the subscribe message.
@@ -231,6 +243,16 @@ func handleClientMessage(ctx context.Context, client *hub.Client, reader api.Rea
 			log.Printf("ws[%s]: failed to send unsubscribed reply: %v", connID, err)
 		}
 
+	case "configure":
+		h.SetResolvePath(client, msg.ResolvePath)
+		reply, _ := json.Marshal(map[string]any{
+			"v": 1, "type": "configured", "id": msg.ID, "resolvePath": msg.ResolvePath,
+		})
+		log.Printf("ws[%s]: configured resolvePath=%t", connID, msg.ResolvePath)
+		if err := conn.Write(ctx, websocket.MessageText, reply); err != nil {
+			log.Printf("ws[%s]: failed to send configured reply: %v", connID, err)
+		}
+
 	case "ping":
 		reply, _ := json.Marshal(map[string]any{"v": 1, "type": "pong", "id": msg.ID})
 		err := conn.Write(ctx, websocket.MessageText, reply)