fix(ws): keep the newest event on overflow and apply route/observer filters

A full send buffer dropped the oldest event and then the new one too while
reporting one drop. Subscribe accepted routeTypes and observerIds but never
applied them.

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

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}) {

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 {

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})
 }

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()