diff --git a/internal/hub/hub.go b/internal/hub/hub.go index 7f67480..87a9ca0 100644 --- a/internal/hub/hub.go +++ b/internal/hub/hub.go @@ -54,17 +54,20 @@ type Event struct { IATA string PayloadType uint8 ChannelHash string // hex string, non-empty only for channelMessage events + RouteType uint8 // packetObservation only + ObserverID string // packetObservation and observerStatus only // Repeat marks a later hearing of a stored observation; only IncludeRepeats clients get it. Repeat bool } // Scope mirrors the client-side subscribe message. All fields are optional: // nil/empty means "no filter on this dimension" (match everything). -// An empty non-nil slice means "match nothing on this dimension". type Scope struct { IATAs []string PayloadTypes []uint8 + RouteTypes []uint8 ChannelHashes []string + ObserverIDs []string Events []EventType } @@ -146,6 +149,12 @@ func scopeMatches(s Scope, e Event) bool { if e.Type == EventChannelMessage && len(s.ChannelHashes) > 0 && !slices.Contains(s.ChannelHashes, e.ChannelHash) { return false } + if e.Type == EventPacketObservation && len(s.RouteTypes) > 0 && !slices.Contains(s.RouteTypes, e.RouteType) { + return false + } + if (e.Type == EventPacketObservation || e.Type == EventObserverStatus) && len(s.ObserverIDs) > 0 && !slices.Contains(s.ObserverIDs, e.ObserverID) { + return false + } return true } @@ -364,12 +373,18 @@ func (h *Hub) Run() { select { case c.Send <- outEvt: default: + // Evict the oldest so the newest still goes out. dropped := 1 select { case <-c.Send: default: } select { + case c.Send <- outEvt: + default: + dropped++ + } + select { case c.laggedCH <- LaggedNotification{DroppedCount: dropped}: default: // laggedCh itself full; write pump will catch up on next drain diff --git a/internal/hub/hub_test.go b/internal/hub/hub_test.go index bb8fc7a..dbaaf0e 100644 --- a/internal/hub/hub_test.go +++ b/internal/hub/hub_test.go @@ -240,6 +240,62 @@ func TestHub_Broadcast_FullBuffer_SendsLaggedNotification(t *testing.T) { } } +func TestScopeMatches_RouteTypeFilter(t *testing.T) { + s := Scope{RouteTypes: []uint8{1}} + if !scopeMatches(s, Event{Type: EventPacketObservation, RouteType: 1}) { + t.Error("expected flood observation to match") + } + if scopeMatches(s, Event{Type: EventPacketObservation, RouteType: 2}) { + t.Error("expected direct observation not to match") + } + if !scopeMatches(s, Event{Type: EventNodeUpdate}) { + t.Error("routeTypes should only filter packetObservation") + } +} + +func TestScopeMatches_ObserverFilter(t *testing.T) { + s := Scope{ObserverIDs: []string{"obs-a"}} + for _, typ := range []EventType{EventPacketObservation, EventObserverStatus} { + if !scopeMatches(s, Event{Type: typ, ObserverID: "obs-a"}) { + t.Errorf("%s from obs-a should match", typ) + } + if scopeMatches(s, Event{Type: typ, ObserverID: "obs-b"}) { + t.Errorf("%s from obs-b should not match", typ) + } + } + if !scopeMatches(s, Event{Type: EventChannelMessage}) { + t.Error("observerIds should not filter channelMessage") + } +} + +func TestHub_Broadcast_FullBuffer_KeepsNewestEvent(t *testing.T) { + h := runHub(t) + c := h.NewClient() + h.AddScope(c, "sub1", Scope{Events: []EventType{EventPacketObservation}}) + time.Sleep(10 * time.Millisecond) + + for i := 0; i < cap(c.Send); i++ { + h.Broadcast(Event{Type: EventPacketObservation, IATA: "YVR", Payload: json.RawMessage(`"old"`)}) + } + h.Broadcast(Event{Type: EventPacketObservation, IATA: "YVR", Payload: json.RawMessage(`"new"`)}) + + select { + case notif := <-c.LaggedCH(): + if notif.DroppedCount != 1 { + t.Errorf("DroppedCount = %d, want 1", notif.DroppedCount) + } + case <-time.After(200 * time.Millisecond): + t.Fatal("expected lagged notification, timed out") + } + var last Event + for len(c.Send) > 0 { + last = <-c.Send + } + if string(last.Payload) != `"new"` { + t.Errorf("newest queued event = %s, want the event that overflowed the buffer", last.Payload) + } +} + func TestClientMatches_NoSubscriptions(t *testing.T) { c := &Client{subscriptions: make(map[string]Scope)} if c.matches(Event{Type: EventPacketObservation}) { diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go index cf00b0c..6c3abb7 100644 --- a/internal/ingest/ingest.go +++ b/internal/ingest/ingest.go @@ -453,6 +453,8 @@ func (w *Worker) broadcastPacketObservation(iata string, payloadType uint8, evt Payload: base, IATA: iata, PayloadType: payloadType, + RouteType: evt.Packet.RouteType, + ObserverID: evt.Observation.ObserverID, } evt.Observation.ResolvedPath = resolvedPath if out.PayloadResolved, err = json.Marshal(evt); err != nil { diff --git a/internal/ingest/status.go b/internal/ingest/status.go index 2e6d8cd..9d3222e 100644 --- a/internal/ingest/status.go +++ b/internal/ingest/status.go @@ -315,5 +315,5 @@ func (w *Worker) handleStatus(ctx context.Context, pubkeyHex string, raw []byte) w.log.Error(fmt.Sprintf("failed to marshal status event payload for %s", pubkeyHex), "error", err) return } - w.hub.Broadcast(hub.Event{Type: hub.EventObserverStatus, Payload: payload, IATA: iata}) + w.hub.Broadcast(hub.Event{Type: hub.EventObserverStatus, Payload: payload, IATA: iata, ObserverID: evt.ObserverID}) } diff --git a/internal/ws/handler.go b/internal/ws/handler.go index 4e93995..9cd27bf 100644 --- a/internal/ws/handler.go +++ b/internal/ws/handler.go @@ -9,7 +9,7 @@ // // Client → Server: // subscribe { v, type, id, scope } → server replies subscribed { v, type, id, subscriptionId } -// unsubscribe { v, type, id, subscriptionId } +// unsubscribe { v, type, id, subscriptionId } → server replies unsubscribed { v, type, id, subscriptionId } // configure { v, type, id, resolvePath, includeObserverKey, includeRepeats } // → server replies configured { v, type, id, resolvePath, includeObserverKey, includeRepeats } // ping { v, type, id } → server replies pong { v, type, id } @@ -25,7 +25,6 @@ // Server → Client events (unsolicited): // packetObservation, observerStatus, nodeUpdate, channelMessage // lagged { v, type, droppedCount, since } -// error { v, type, code, message } // // Idle connections (no ping) closed after 90s. // Client should ping every 30s. @@ -235,7 +234,9 @@ func handleClientMessage(ctx context.Context, client *hub.Client, reader api.Rea scope := hub.Scope{ IATAs: iatas, PayloadTypes: msg.Scope.PayloadTypes, + RouteTypes: msg.Scope.RouteTypes, ChannelHashes: msg.Scope.ChannelHashes, + ObserverIDs: msg.Scope.ObserverIDs, Events: msg.Scope.Events, } subID := uuid.NewString()