feat(tags): wire tag-enrichment worker at startup (#1490 wiring)
Construct the tag SettingsService + Enricher at boot (mirroring coverart: reconcile providers, bump the sources version if the provider set changed to re-open settled rows), then run a standalone background Worker that drains tracks needing folksonomy tags on a periodic tick. Standalone (not threaded through the file-scan chain like cover art) because tag lookups need only DB fields — recording MBID / artist / title — so it mirrors the ListenBrainz similarity worker instead: an initial drain shortly after boot, then every 30 min, up to 200 tracks per tick. MusicBrainz's 1 req/s ceiling is the real throttle. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -32,6 +32,7 @@ import (
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/similarity"
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/subsonic"
|
||||
syncpkg "git.fabledsword.com/bvandeusen/minstrel/internal/sync"
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/tags"
|
||||
)
|
||||
|
||||
func main() {
|
||||
@@ -205,6 +206,25 @@ func run() error {
|
||||
similarityWorker := similarity.NewWorker(pool, listenbrainz.NewClient(), logger.With("component", "similarity"))
|
||||
go similarityWorker.Run(ctx)
|
||||
|
||||
// Start the tag-enrichment worker (#1490). Reconciles the compiled-in
|
||||
// tag providers with tag_provider_settings, bumps the sources version if
|
||||
// the provider set changed (re-opening settled rows), then drains tracks
|
||||
// needing folksonomy tags on a periodic tick. Standalone (not in the file
|
||||
// scan chain) because tag lookups need only DB fields — MBID / artist /
|
||||
// title — that a scan has already imported.
|
||||
tagSettings, err := tags.NewSettingsService(ctx, pool, logger.With("component", "tags"))
|
||||
if err != nil {
|
||||
logger.Error("tag settings service init failed", "err", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
if newVer, bumped, berr := tagSettings.BumpVersionIfProvidersChanged(ctx); berr != nil {
|
||||
logger.Warn("tags: provider-hash boot check failed", "err", berr)
|
||||
} else if bumped {
|
||||
logger.Info("tags: registered provider set changed; version bumped", "new_version", newVer)
|
||||
}
|
||||
tagEnricher := tags.NewEnricher(pool, logger.With("component", "tags"), tagSettings)
|
||||
go tags.NewWorker(tagEnricher, logger.With("component", "tags")).Run(ctx)
|
||||
|
||||
// Start the GC worker. Runs every 1h and sweeps lifecycle tables
|
||||
// that have no writer-side close path or retention policy:
|
||||
// orphan play_events, stale play_sessions, expired
|
||||
|
||||
@@ -0,0 +1,59 @@
|
||||
package tags
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Worker periodically drains tracks needing tag enrichment. Unlike cover
|
||||
// art (threaded through the file-scan chain because it needs track file
|
||||
// paths), tag enrichment only needs data already in the DB — recording
|
||||
// MBID / artist / title — so it runs as a standalone background worker,
|
||||
// mirroring the ListenBrainz similarity worker. Rate limiting lives in the
|
||||
// providers' httpClients, so a tick just drains a bounded batch and the
|
||||
// external APIs pace themselves.
|
||||
type Worker struct {
|
||||
enricher *Enricher
|
||||
logger *slog.Logger
|
||||
tick time.Duration
|
||||
batch int
|
||||
}
|
||||
|
||||
// NewWorker constructs a worker with production defaults: an initial drain
|
||||
// shortly after boot, then every 30 minutes, up to 200 tracks per tick.
|
||||
// MusicBrainz's 1 req/s ceiling is the real throttle, so the batch size
|
||||
// mainly bounds how long one tick runs, not the request rate.
|
||||
func NewWorker(enricher *Enricher, logger *slog.Logger) *Worker {
|
||||
return &Worker{
|
||||
enricher: enricher,
|
||||
logger: logger,
|
||||
tick: 30 * time.Minute,
|
||||
batch: 200,
|
||||
}
|
||||
}
|
||||
|
||||
// Run blocks until ctx is cancelled: an initial drain, then every w.tick.
|
||||
func (w *Worker) Run(ctx context.Context) {
|
||||
w.tickOnce(ctx)
|
||||
t := time.NewTicker(w.tick)
|
||||
defer t.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-t.C:
|
||||
w.tickOnce(ctx)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// tickOnce drains one bounded batch. EnrichTrackBatch already logs a
|
||||
// category breakdown, so this only surfaces a fatal batch error.
|
||||
func (w *Worker) tickOnce(ctx context.Context) {
|
||||
if _, _, _, err := w.enricher.EnrichTrackBatch(ctx, w.batch, nil); err != nil {
|
||||
if ctx.Err() == nil {
|
||||
w.logger.Error("tags: enrichment tick failed", "err", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user