feat: sse

go.mod

@@ -24,6 +24,7 @@ require (
 	github.com/spf13/cobra v1.10.1
 	github.com/spf13/viper v1.21.0
 	github.com/stretchr/testify v1.11.1
+	github.com/tmaxmax/go-sse v0.11.0
 	github.com/zeebo/bencode v1.0.0
 	golang.org/x/crypto v0.43.0
 	golang.org/x/image v0.32.0

go.sum

@@ -158,6 +158,8 @@ github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu
 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
 github.com/subosito/gotenv v1.6.0 h1:9NlTDc1FTs4qu0DDq7AEtTPNw6SVm7uBMsUCUjABIf8=
 github.com/subosito/gotenv v1.6.0/go.mod h1:Dk4QP5c2W3ibzajGcXpNraDfq2IrhjMIvMSWPKKo0FU=
+github.com/tmaxmax/go-sse v0.11.0 h1:nogmJM6rJUoOLoAwEKeQe5XlVpt9l7N82SS1jI7lWFg=
+github.com/tmaxmax/go-sse v0.11.0/go.mod h1:u/2kZQR1tyngo1lKaNCj1mJmhXGZWS1Zs5yiSOD+Eg8=
 github.com/ulikunitz/xz v0.5.11/go.mod h1:nbz6k7qbPmH4IRqmfOplQw/tblSgqTqBwxkY0oWt/14=
 github.com/ulikunitz/xz v0.5.15 h1:9DNdB5s+SgV3bQ2ApL10xRc35ck0DuIX/isZvIk+ubY=
 github.com/ulikunitz/xz v0.5.15/go.mod h1:nbz6k7qbPmH4IRqmfOplQw/tblSgqTqBwxkY0oWt/14=

internal/api/server.go

@@ -21,6 +21,7 @@ import (
 
 	"github.com/autobrr/qui/internal/api/handlers"
 	"github.com/autobrr/qui/internal/api/middleware"
+	"github.com/autobrr/qui/internal/api/sse"
 	"github.com/autobrr/qui/internal/auth"
 	"github.com/autobrr/qui/internal/backups"
 	"github.com/autobrr/qui/internal/config"
@@ -51,6 +52,7 @@ type Server struct {
 	updateService      *update.Service
 	trackerIconService *trackericons.Service
 	backupService      *backups.Service
+	streamManager      *sse.StreamManager
 }
 
 type Dependencies struct {
@@ -70,6 +72,11 @@ type Dependencies struct {
 }
 
 func NewServer(deps *Dependencies) *Server {
+	streamManager := sse.NewStreamManager(deps.ClientPool, deps.SyncManager, deps.InstanceStore)
+	if deps.ClientPool != nil {
+		deps.ClientPool.SetSyncEventSink(streamManager)
+	}
+
 	s := Server{
 		server: &http.Server{
 			ReadHeaderTimeout: time.Second * 15,
@@ -90,6 +97,7 @@ func NewServer(deps *Dependencies) *Server {
 		updateService:      deps.UpdateService,
 		trackerIconService: deps.TrackerIconService,
 		backupService:      deps.BackupService,
+		streamManager:      streamManager,
 	}
 
 	return &s
@@ -152,6 +160,15 @@ func (s *Server) tryToServe(addr, protocol string) error {
 }
 
 func (s *Server) Shutdown(ctx context.Context) error {
+	if s.streamManager != nil {
+		shutdownCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
+		defer cancel()
+
+		if err := s.streamManager.Shutdown(shutdownCtx); err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
+			s.logger.Warn().Err(err).Msg("failed to shut down stream manager cleanly")
+		}
+	}
+
 	return s.server.Shutdown(ctx)
 }
 
@@ -300,6 +317,9 @@ func (s *Server) Handler() (*chi.Mux, error) {
 					})
 
 					r.Get("/capabilities", instancesHandler.GetInstanceCapabilities)
+					if s.streamManager != nil {
+						r.Get("/stream", s.streamManager.ServeInstance)
+					}
 
 					// Torrent creator
 					r.Route("/torrent-creator", func(r chi.Router) {

internal/api/server_test.go

@@ -9,9 +9,11 @@ import (
 	"fmt"
 	"net/http"
 	"path/filepath"
+	"reflect"
 	"sort"
 	"strings"
 	"testing"
+	"unsafe"
 
 	"github.com/alexedwards/scs/v2"
 	"github.com/go-chi/chi/v5"
@@ -39,6 +41,7 @@ type routeKey struct {
 
 var undocumentedRoutes = map[routeKey]struct{}{
 	{Method: http.MethodGet, Path: "/api/auth/validate"}:                                                            {},
+	{Method: http.MethodGet, Path: "/api/instances/{instanceId}/stream"}:                                            {},
 	{Method: http.MethodPost, Path: "/api/instances/{instanceId}/backups/run"}:                                      {},
 	{Method: http.MethodGet, Path: "/api/instances/{instanceId}/backups/runs"}:                                      {},
 	{Method: http.MethodDelete, Path: "/api/instances/{instanceId}/backups/runs"}:                                   {},
@@ -50,6 +53,21 @@ var undocumentedRoutes = map[routeKey]struct{}{
 	{Method: http.MethodPut, Path: "/api/instances/{instanceId}/backups/settings"}:                                  {},
 }
 
+func TestNewServerRegistersStreamManagerAsSyncSink(t *testing.T) {
+	clientPool := &qbittorrent.ClientPool{}
+
+	server := NewServer(&Dependencies{
+		Config:     &config.AppConfig{Config: &domain.Config{BaseURL: "/"}},
+		ClientPool: clientPool,
+	})
+
+	require.NotNil(t, server.streamManager, "expected stream manager to be initialized")
+
+	sink := getClientPoolSyncEventSink(t, clientPool)
+	require.NotNil(t, sink, "expected client pool to have a sync sink registered")
+	require.Same(t, server.streamManager, sink, "stream manager should be registered as sync sink")
+}
+
 func TestAllEndpointsDocumented(t *testing.T) {
 	server := NewServer(newTestDependencies(t))
 	router, err := server.Handler()
@@ -243,3 +261,21 @@ func formatRoutes(routes []routeKey) string {
 	}
 	return strings.Join(lines, "\n")
 }
+
+func getClientPoolSyncEventSink(t *testing.T, pool *qbittorrent.ClientPool) qbittorrent.SyncEventSink {
+	t.Helper()
+
+	value := reflect.ValueOf(pool).Elem().FieldByName("syncEventSink")
+	if !value.IsValid() {
+		t.Fatalf("client pool does not expose syncEventSink field")
+	}
+
+	exposed := reflect.NewAt(value.Type(), unsafe.Pointer(value.UnsafeAddr())).Elem()
+	if exposed.IsNil() {
+		return nil
+	}
+
+	sink, ok := exposed.Interface().(qbittorrent.SyncEventSink)
+	require.True(t, ok, "unexpected sink type stored on client pool")
+	return sink
+}