feat(service): add peer observer registry and middleware

Adds an in-process observer that records device->service requests by
source IP. PeerObserverMiddleware fires on every inbound after RealIP
trust and Recoverer; the registry exposes Register/Signal/Forget keyed
on the device IP with a buffered one-shot delivery.

No callers yet — this is the substrate for the passive reachability
probe that replaces the broken active swUpdateUrl round-trip on
migrated speakers.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

cmd/soundtouch-service/main.go

@@ -865,6 +865,7 @@ func setupRouter(server *handlers.Server) *chi.Mux {
 	r.Use(server.SnapshotMiddleware)
 	r.Use(server.OriginMiddleware)
 	r.Use(middleware.Recoverer)
+	r.Use(server.PeerObserverMiddleware)
 	r.Use(server.ShortcutMiddleware)
 	r.Use(server.MirrorMiddleware)
 	r.Use(server.RecordMiddleware)

pkg/service/handlers/middleware_peer_observer.go

@@ -0,0 +1,29 @@
+package handlers
+
+import (
+	"net"
+	"net/http"
+	"time"
+)
+
+// PeerObserverMiddleware records every incoming request's source IP and
+// path in the peerObserver registry. It fires on every request before
+// the handler runs, so passive reachability probes can register a device
+// IP and learn whether any inbound landed in their wait window.
+//
+// Placement: after TrustedRealIPMiddleware (so r.RemoteAddr reflects the
+// trusted client IP) and after Recoverer (so any panic inside this
+// middleware is contained). Before any short-circuiting middleware
+// would be unnecessary — Signal runs before next.ServeHTTP, so the
+// observation lands regardless of how later middleware handles the
+// request.
+func (s *Server) PeerObserverMiddleware(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		host, _, err := net.SplitHostPort(r.RemoteAddr)
+		if err == nil && host != "" {
+			s.peerObserver.Signal(host, PeerHit{Path: r.URL.Path, At: time.Now()})
+		}
+
+		next.ServeHTTP(w, r)
+	})
+}

pkg/service/handlers/peer_observer.go

@@ -0,0 +1,76 @@
+package handlers
+
+import (
+	"sync"
+	"time"
+)
+
+// PeerHit is the payload delivered to a waiter when a request from a
+// registered peer IP lands on the service. Path is the chi URL path of
+// the inbound; At is the wall-clock time the middleware saw it.
+type PeerHit struct {
+	Path string
+	At   time.Time
+}
+
+// peerObserver is the rendezvous between the passive reachability probe
+// (which registers interest in a device IP and waits for any inbound)
+// and the chi middleware (which signals on every request whose source
+// IP matches a registration).
+//
+// Unlike probeRegistry, which keys on a unique per-probe token, this
+// observer keys on the device's IP — the probe doesn't mutate device
+// state, so there's no token to thread through the request path. Any
+// inbound from the IP counts as proof of reachability.
+type peerObserver struct {
+	mu      sync.Mutex
+	pending map[string]chan PeerHit
+}
+
+func newPeerObserver() *peerObserver {
+	return &peerObserver{pending: make(map[string]chan PeerHit)}
+}
+
+// Register creates a one-shot buffered channel keyed by IP. The buffer
+// of 1 lets the middleware deliver the first hit and silently drop
+// subsequent hits during the wait window without blocking. Caller is
+// responsible for pairing every Register with Forget.
+func (o *peerObserver) Register(ip string) <-chan PeerHit {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+
+	ch := make(chan PeerHit, 1)
+	o.pending[ip] = ch
+
+	return ch
+}
+
+// Signal delivers a hit to the channel for ip, non-blocking. Returns
+// true when a matching registration existed AND the hit was delivered
+// (i.e. the channel had buffer space — first hit during the window).
+// Subsequent hits during the same window return false without blocking.
+func (o *peerObserver) Signal(ip string, hit PeerHit) bool {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+
+	ch, ok := o.pending[ip]
+	if !ok {
+		return false
+	}
+
+	select {
+	case ch <- hit:
+		return true
+	default:
+		return false
+	}
+}
+
+// Forget removes the entry. Safe to call regardless of whether a hit
+// landed — does not affect already-returned channels.
+func (o *peerObserver) Forget(ip string) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+
+	delete(o.pending, ip)
+}

pkg/service/handlers/peer_observer_test.go

@@ -0,0 +1,69 @@
+package handlers
+
+import (
+	"testing"
+	"time"
+)
+
+func TestPeerObserver_RegisterSignalForget(t *testing.T) {
+	o := newPeerObserver()
+
+	ch := o.Register("192.168.1.42")
+	if ch == nil {
+		t.Fatal("Register returned nil channel")
+	}
+
+	first := PeerHit{Path: "/updates/soundtouch", At: time.Now()}
+	if !o.Signal("192.168.1.42", first) {
+		t.Error("Signal returned false for registered IP")
+	}
+
+	// Second signal while the buffer is still full (no reader yet) drops
+	// silently and returns false — only the first hit per window matters.
+	if o.Signal("192.168.1.42", PeerHit{Path: "/streaming/x"}) {
+		t.Error("second Signal returned true; expected false (buffer full, undrained)")
+	}
+
+	select {
+	case got := <-ch:
+		if got.Path != first.Path {
+			t.Errorf("hit.Path = %q, want %q", got.Path, first.Path)
+		}
+	case <-time.After(100 * time.Millisecond):
+		t.Error("Signal did not deliver hit to channel")
+	}
+
+	o.Forget("192.168.1.42")
+
+	// After Forget, Signal returns false.
+	if o.Signal("192.168.1.42", first) {
+		t.Error("Signal returned true after Forget")
+	}
+}
+
+func TestPeerObserver_UnknownIP(t *testing.T) {
+	o := newPeerObserver()
+	if o.Signal("10.0.0.1", PeerHit{Path: "/anything"}) {
+		t.Error("Signal returned true for unregistered IP")
+	}
+}
+
+func TestPeerObserver_SignalIsNonBlocking(t *testing.T) {
+	o := newPeerObserver()
+	o.Register("192.168.1.42") // never drain
+
+	done := make(chan struct{})
+	go func() {
+		for i := 0; i < 100; i++ {
+			o.Signal("192.168.1.42", PeerHit{Path: "/x"})
+		}
+		close(done)
+	}()
+
+	select {
+	case <-done:
+		// Signal never blocked even with no reader and a full buffer.
+	case <-time.After(500 * time.Millisecond):
+		t.Fatal("Signal blocked when buffer was full — must drop silently")
+	}
+}

pkg/service/handlers/server.go

@@ -62,6 +62,7 @@ type Server struct {
 	amazonRedirectURI   string
 	amazonService       *amazon.Service
 	probes              *probeRegistry
+	peerObserver        *peerObserver
 }
 
 // RequestSnapshot represents an immutable snapshot of an HTTP request.
@@ -97,6 +98,7 @@ func NewServer(ds *datastore.DataStore, sm *setup.Manager, serverURL string, pro
 		discoveryInterval: 5 * time.Minute,
 		discoveryEnabled:  true,
 		probes:            newProbeRegistry(),
+		peerObserver:      newPeerObserver(),
 	}
 
 	return s