feat(similarity): add Worker skeleton with constructor + Run loop
This commit is contained in:
@@ -0,0 +1,66 @@
|
||||
// Package similarity owns the inbound ListenBrainz similarity ingest
|
||||
// pipeline. A periodic worker queries LB's /explore/similar-recordings
|
||||
// and /explore/similar-artists endpoints for tracks the user has played,
|
||||
// filters returned MBIDs to the local library, and stores the top-K
|
||||
// edges in track_similarity / artist_similarity for M4c's radio
|
||||
// candidate-pool builder.
|
||||
package similarity
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/scrobble/listenbrainz"
|
||||
)
|
||||
|
||||
// Worker drains played-tracks-and-artists needing similarity and POSTs
|
||||
// the results into track_similarity / artist_similarity. Failures are
|
||||
// passively retried via the timer (no durable queue table — losing one
|
||||
// tick's worth of refresh attempts is "1 hour of staleness," fine).
|
||||
type Worker struct {
|
||||
pool *pgxpool.Pool
|
||||
client *listenbrainz.Client
|
||||
logger *slog.Logger
|
||||
tick time.Duration
|
||||
batch int32
|
||||
topK int
|
||||
}
|
||||
|
||||
// NewWorker constructs a worker with production defaults: 1h tick,
|
||||
// batch=5, topK=20.
|
||||
func NewWorker(pool *pgxpool.Pool, client *listenbrainz.Client, logger *slog.Logger) *Worker {
|
||||
return &Worker{
|
||||
pool: pool,
|
||||
client: client,
|
||||
logger: logger,
|
||||
tick: 1 * time.Hour,
|
||||
batch: 5,
|
||||
topK: 20,
|
||||
}
|
||||
}
|
||||
|
||||
// Run blocks until ctx is cancelled, ticking every w.tick.
|
||||
func (w *Worker) Run(ctx context.Context) {
|
||||
t := time.NewTicker(w.tick)
|
||||
defer t.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-t.C:
|
||||
if err := w.tickOnce(ctx); err != nil {
|
||||
w.logger.Error("similarity: tick failed", "err", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// tickOnce drains one batch of tracks and one batch of artists. Stub —
|
||||
// implementation lands in Task 6 along with the integration tests that
|
||||
// drive it.
|
||||
func (w *Worker) tickOnce(_ context.Context) error {
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,22 @@
|
||||
package similarity
|
||||
|
||||
import (
|
||||
"io"
|
||||
"log/slog"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestNewWorker_DefaultsMatchSpec(t *testing.T) {
|
||||
logger := slog.New(slog.NewTextHandler(io.Discard, nil))
|
||||
w := NewWorker(nil, nil, logger)
|
||||
if w.tick != 1*time.Hour {
|
||||
t.Errorf("tick = %v, want 1h", w.tick)
|
||||
}
|
||||
if w.batch != 5 {
|
||||
t.Errorf("batch = %d, want 5", w.batch)
|
||||
}
|
||||
if w.topK != 20 {
|
||||
t.Errorf("topK = %d, want 20", w.topK)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user