diff --git a/README.md b/README.md index 9d620bb..699875c 100644 --- a/README.md +++ b/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 diff --git a/internal/hub/hub.go b/internal/hub/hub.go index 4184107..3acc5b2 100644 --- a/internal/hub/hub.go +++ b/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 { diff --git a/internal/hub/hub_test.go b/internal/hub/hub_test.go index face697..d2bee3f 100644 --- a/internal/hub/hub_test.go +++ b/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) + } +} diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go index a891346..fd703e8 100644 --- a/internal/ingest/ingest.go +++ b/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. diff --git a/internal/ingest/packet.go b/internal/ingest/packet.go index 30dd97b..92fec21 100644 --- a/internal/ingest/packet.go +++ b/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) } } diff --git a/internal/ws/handler.go b/internal/ws/handler.go index 2faa0e1..c21826e 100644 --- a/internal/ws/handler.go +++ b/internal/ws/handler.go @@ -8,9 +8,16 @@ // On connect: server sends hello { v:1, type:"hello", serverTime:, 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)