diff --git a/cmd/soundtouch-service/main.go b/cmd/soundtouch-service/main.go index c041e04..25bc3ac 100644 --- a/cmd/soundtouch-service/main.go +++ b/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) diff --git a/pkg/service/handlers/middleware_peer_observer.go b/pkg/service/handlers/middleware_peer_observer.go new file mode 100644 index 0000000..8df79a9 --- /dev/null +++ b/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) + }) +} \ No newline at end of file diff --git a/pkg/service/handlers/peer_observer.go b/pkg/service/handlers/peer_observer.go new file mode 100644 index 0000000..9fbb2a8 --- /dev/null +++ b/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) +} \ No newline at end of file diff --git a/pkg/service/handlers/peer_observer_test.go b/pkg/service/handlers/peer_observer_test.go new file mode 100644 index 0000000..8cbdbde --- /dev/null +++ b/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") + } +} \ No newline at end of file diff --git a/pkg/service/handlers/server.go b/pkg/service/handlers/server.go index 08e6e5a..01ad994 100644 --- a/pkg/service/handlers/server.go +++ b/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