From b5849d6ad142af38a14732f3395fb338c13651c4 Mon Sep 17 00:00:00 2001 From: Alexander Date: Wed, 20 May 2026 14:57:02 +0200 Subject: [PATCH] Wire shared EventBus from main into PollDownloadWorker and server Create EventBus in serveGrpc() and pass to both setupRiver() and NewMusicAgregatorServer() so PollDownloadWorker and the gRPC server share the same event bus instance. Ultraworked with [Sisyphus](https://github.com/code-yeongyu/claude-agent) Co-authored-by: Sisyphus --- cmd/music-agregator/main.go | 11 ++++++++--- internal/server.go | 3 +-- 2 files changed, 9 insertions(+), 5 deletions(-) diff --git a/cmd/music-agregator/main.go b/cmd/music-agregator/main.go index e86826e..5458c2f 100644 --- a/cmd/music-agregator/main.go +++ b/cmd/music-agregator/main.go @@ -29,6 +29,7 @@ import ( "homelab.lan/music-agregator/internal/analysis" "homelab.lan/music-agregator/internal/config" "homelab.lan/music-agregator/internal/database" + "homelab.lan/music-agregator/internal/eventbus" "homelab.lan/music-agregator/internal/hello" "homelab.lan/music-agregator/internal/indexer" "homelab.lan/music-agregator/internal/metadata" @@ -84,7 +85,7 @@ type riverSetup struct { cacheRefreshWorker *indexer.CacheRefreshWorker } -func setupRiver(ctx context.Context, cfg config.Config, db *database.DB, torrentClient torrent.TorrentClient, pathMapper *torrent.PathMapper, musicfsClient *musicfs.Client, metadataService *metadata.MetadataService) *riverSetup { +func setupRiver(ctx context.Context, cfg config.Config, db *database.DB, torrentClient torrent.TorrentClient, pathMapper *torrent.PathMapper, musicfsClient *musicfs.Client, metadataService *metadata.MetadataService, bus *eventbus.EventBus) *riverSetup { cacheWorker := &indexer.CacheRefreshWorker{} pollWorker := &workers.PollDownloadWorker{ Downloads: database.NewDownloadRepository(db.Pool), @@ -98,6 +99,8 @@ func setupRiver(ctx context.Context, cfg config.Config, db *database.DB, torrent MusicFSOriginID: cfg.MusicFS.OriginID, MusicFSOriginRoot: cfg.MusicFS.OriginRoot, MetadataService: metadataService, + EventBus: bus, + AlbumEvents: database.NewAlbumEventRepository(db.Pool), } riverWorkers := river.NewWorkers() @@ -194,9 +197,11 @@ func serveGrpc(config config.Config) { defer metadataConn.Close() metadataService := metadata.NewMetadataService(metadataClient, db) - rs := setupRiver(ctx, config, db, torrentClient, pathMapper, musicfsClient, metadataService) + bus := eventbus.New() - musiscAgregatorSeerver, err := internal.NewMusicAgregatorServer(config, rs.client, torrentClient, pathMapper, db) + rs := setupRiver(ctx, config, db, torrentClient, pathMapper, musicfsClient, metadataService, bus) + + musiscAgregatorSeerver, err := internal.NewMusicAgregatorServer(config, rs.client, torrentClient, pathMapper, db, bus) if err != nil { log.Fatal().Err(err).Msg("failed to create MusicAgregatorServer") } diff --git a/internal/server.go b/internal/server.go index 2686b8a..55906b4 100644 --- a/internal/server.go +++ b/internal/server.go @@ -26,13 +26,12 @@ type MusicAgregatorServer struct { pb.UnimplementedMusicAgregatorServiceServer } -func NewMusicAgregatorServer(cfg config.Config, riverClient *river.Client[pgx.Tx], torrentClient torrent.TorrentClient, pathMapper *torrent.PathMapper, db *database.DB) (*MusicAgregatorServer, error) { +func NewMusicAgregatorServer(cfg config.Config, riverClient *river.Client[pgx.Tx], torrentClient torrent.TorrentClient, pathMapper *torrent.PathMapper, db *database.DB, bus *eventbus.EventBus) (*MusicAgregatorServer, error) { service, err := NewMusicAgregatorService(cfg, riverClient, torrentClient, pathMapper, db) if err != nil { log.Err(err).Msg("failed to create MusicAgregatorService") return nil, err } - bus := eventbus.New() return &MusicAgregatorServer{ service: service, bus: bus,