fujin/metadata-agregator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: fuzzy search with parallel DB/MusicBrainz execution

Browse Source 
- Add pg_trgm fuzzy search to PostgreSQL artist repository
- Add Lucene fuzzy query (~0.7) for MusicBrainz artist search
- Run DB and MusicBrainz searches in parallel goroutines
- Cache MusicBrainz results to DB for accumulation
- Deduplicate results by external ID
- Filter out special purpose artists (type=Other)
- Add SaveAll method for batch artist inserts
- Move database schema to containers repo
This commit is contained in:
Alexander
2026-05-10 21:13:17 +02:00
parent b4e763be44
commit 2756cc1043
9 changed files with 277 additions and 263 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/provider/musicbrainz/provider.go
+27 -2
View File
@@ -42,8 +42,8 @@ func (p *Provider) SearchArtists(ctx context.Context, query string, limit, offse
limit = 25
}
escapedQuery := escapeQuery(query)
data, err := p.client.search(ctx, "artist", fmt.Sprintf("artist:%s", escapedQuery), limit, offset)
luceneQuery := buildFuzzyArtistQuery(query, 0.7)
data, err := p.client.search(ctx, "artist", luceneQuery, limit, offset)
if err != nil {
return nil, fmt.Errorf("search artists: %w", err)
}
@@ -311,6 +311,31 @@ func selectCanonicalRelease(releases []*mbRelease) *mbRelease {
return best
}
func buildFuzzyArtistQuery(query string, similarity float64) string {
terms := strings.Fields(query)
if len(terms) == 0 {
return ""
}
var parts []string
for _, term := range terms {
escaped := escapeForFuzzy(term)
parts = append(parts, fmt.Sprintf("%s~%.1f", escaped, similarity))
}
termQuery := strings.Join(parts, " AND ")
return fmt.Sprintf("(artist:(%s) OR alias:(%s))", termQuery, termQuery)
}
func escapeForFuzzy(s string) string {
special := []string{`+`, `-`, `&`, `|`, `!`, `(`, `)`, `{`, `}`, `[`, `]`, `^`, `"`, `:`, `/`, `\`}
result := s
for _, char := range special {
result = strings.ReplaceAll(result, char, `\`+char)
}
return result
}
func escapeQuery(s string) string {
special := []string{`+`, `-`, `&`, `|`, `!`, `(`, `)`, `{`, `}`, `[`, `]`, `^`, `"`, `~`, `*`, `?`, `:`, `/`, `\`}
result := s
internal/repository/postgres/artist.go
+129 -28
View File
@@ -60,23 +60,31 @@ func (r *ArtistRepository) GetByExternalID(ctx context.Context, source, sourceID
}
func (r *ArtistRepository) Search(ctx context.Context, query string, limit, offset int) (*domain.SearchResult[domain.Artist], error) {
countQuery := `SELECT COUNT(*) FROM artists WHERE name ILIKE $1`
// Use pg_trgm similarity for fuzzy matching with a threshold of 0.3
// Combined with ILIKE for substring matches
countQuery := `
SELECT COUNT(*) FROM artists
WHERE name ILIKE $1 OR similarity(lower(name), lower($2)) > 0.3`
searchQuery := `
SELECT id, name, sort_name, artist_type, country, formed_date, disbanded_date,
description, image_url, source, source_id
description, image_url, source, source_id,
GREATEST(
similarity(lower(name), lower($2)),
CASE WHEN name ILIKE $1 THEN 0.5 ELSE 0 END
) as score
FROM artists
WHERE name ILIKE $1
ORDER BY name
LIMIT $2 OFFSET $3`
WHERE name ILIKE $1 OR similarity(lower(name), lower($2)) > 0.3
ORDER BY score DESC, name
LIMIT $3 OFFSET $4`
pattern := "%" + query + "%"
var total int
if err := r.pool.QueryRow(ctx, countQuery, pattern).Scan(&total); err != nil {
if err := r.pool.QueryRow(ctx, countQuery, pattern, query).Scan(&total); err != nil {
return nil, err
}
rows, err := r.pool.Query(ctx, searchQuery, pattern, limit, offset)
rows, err := r.pool.Query(ctx, searchQuery, pattern, query, limit, offset)
if err != nil {
return nil, err
}
@@ -84,7 +92,7 @@ func (r *ArtistRepository) Search(ctx context.Context, query string, limit, offs
var artists []domain.Artist
for rows.Next() {
artist, err := r.scanArtistFromRow(rows)
artist, err := r.scanArtistFromRowWithScore(rows)
if err != nil {
return nil, err
}
@@ -152,20 +160,79 @@ func (r *ArtistRepository) Save(ctx context.Context, artist *domain.Artist) erro
return tx.Commit(ctx)
}
func (r *ArtistRepository) SaveAll(ctx context.Context, artists []domain.Artist) error {
if len(artists) == 0 {
return nil
}
tx, err := r.pool.Begin(ctx)
if err != nil {
return err
}
defer tx.Rollback(ctx)
for _, artist := range artists {
var source, sourceID string
if len(artist.ExternalIDs) > 0 {
source = artist.ExternalIDs[0].Source
sourceID = artist.ExternalIDs[0].SourceID
}
query := `
INSERT INTO artists (id, name, sort_name, artist_type, country, formed_date,
disbanded_date, description, image_url, source, source_id)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
ON CONFLICT (id) DO UPDATE SET
name = EXCLUDED.name,
sort_name = EXCLUDED.sort_name,
artist_type = EXCLUDED.artist_type,
country = EXCLUDED.country,
formed_date = EXCLUDED.formed_date,
disbanded_date = EXCLUDED.disbanded_date,
description = EXCLUDED.description,
image_url = EXCLUDED.image_url,
updated_at = now()`
_, err = tx.Exec(ctx, query,
artist.ID, artist.Name, nullString(artist.SortName), nullString(artist.Type),
nullString(artist.Country), artist.FormedDate, artist.DisbandedDate,
nullString(artist.Description), nullString(artist.ImageURL), source, sourceID)
if err != nil {
return err
}
for _, ext := range artist.ExternalIDs {
extQuery := `
INSERT INTO artist_external_ids (artist_id, source, source_id, url)
VALUES ($1, $2, $3, $4)
ON CONFLICT (artist_id, source, source_id) DO UPDATE SET
url = EXCLUDED.url,
fetched_at = now()`
_, err = tx.Exec(ctx, extQuery, artist.ID, ext.Source, ext.SourceID, nullString(ext.URL))
if err != nil {
return err
}
}
}
return tx.Commit(ctx)
}
func (r *ArtistRepository) scanArtist(ctx context.Context, query string, args ...any) (*domain.Artist, error) {
row := r.pool.QueryRow(ctx, query, args...)
var (
artist domain.Artist
sortName *string
artistType *string
country *string
formedDate *time.Time
disbandDate *time.Time
description *string
imageURL *string
source string
sourceID *string
artist domain.Artist
sortName *string
artistType *string
country *string
formedDate *time.Time
disbandDate *time.Time
description *string
imageURL *string
source string
sourceID *string
)
err := row.Scan(
@@ -192,16 +259,16 @@ func (r *ArtistRepository) scanArtist(ctx context.Context, query string, args ..
func (r *ArtistRepository) scanArtistFromRow(row pgx.Row) (*domain.Artist, error) {
var (
artist domain.Artist
sortName *string
artistType *string
country *string
formedDate *time.Time
disbandDate *time.Time
description *string
imageURL *string
source string
sourceID *string
artist domain.Artist
sortName *string
artistType *string
country *string
formedDate *time.Time
disbandDate *time.Time
description *string
imageURL *string
source string
sourceID *string
)
err := row.Scan(
@@ -223,6 +290,40 @@ func (r *ArtistRepository) scanArtistFromRow(row pgx.Row) (*domain.Artist, error
return &artist, nil
}
func (r *ArtistRepository) scanArtistFromRowWithScore(row pgx.Row) (*domain.Artist, error) {
var (
artist domain.Artist
sortName *string
artistType *string
country *string
formedDate *time.Time
disbandDate *time.Time
description *string
imageURL *string
source string
sourceID *string
score float64
)
err := row.Scan(
&artist.ID, &artist.Name, &sortName, &artistType, &country,
&formedDate, &disbandDate, &description, &imageURL, &source, &sourceID, &score,
)
if err != nil {
return nil, err
}
artist.SortName = derefString(sortName)
artist.Type = derefString(artistType)
artist.Country = derefString(country)
artist.FormedDate = formedDate
artist.DisbandedDate = disbandDate
artist.Description = derefString(description)
artist.ImageURL = derefString(imageURL)
return &artist, nil
}
func (r *ArtistRepository) loadExternalIDs(ctx context.Context, artist *domain.Artist) error {
query := `SELECT source, source_id, url FROM artist_external_ids WHERE artist_id = $1`
internal/repository/repository.go
+1
View File
@@ -11,6 +11,7 @@ type ArtistRepository interface {
GetByExternalID(ctx context.Context, source, sourceID string) (*domain.Artist, error)
Search(ctx context.Context, query string, limit, offset int) (*domain.SearchResult[domain.Artist], error)
Save(ctx context.Context, artist *domain.Artist) error
SaveAll(ctx context.Context, artists []domain.Artist) error
}
type AlbumRepository interface {
internal/service/metadata.go
+112 -8
View File
@@ -3,6 +3,7 @@ package service
import (
"context"
"errors"
"sync"
"github.com/rs/zerolog"
@@ -68,16 +69,119 @@ func (s *MetadataService) GetArtist(ctx context.Context, id string) (*domain.Art
func (s *MetadataService) SearchArtists(ctx context.Context, query string, limit, offset int) (*domain.SearchResult[domain.Artist], error) {
log := zerolog.Ctx(ctx)
result, err := s.artists.Search(ctx, query, limit, offset)
if err == nil && len(result.Items) > 0 {
log.Debug().Int("count", len(result.Items)).Msg("artist search served from cache")
metrics.CacheHits.WithLabelValues("artist_search").Inc()
return result, nil
var (
wg sync.WaitGroup
dbResult *domain.SearchResult[domain.Artist]
mbResult *domain.SearchResult[domain.Artist]
dbErr error
mbErr error
)
wg.Add(2)
go func() {
defer wg.Done()
dbResult, dbErr = s.artists.Search(ctx, query, limit, offset)
if dbErr == nil && len(dbResult.Items) > 0 {
metrics.CacheHits.WithLabelValues("artist_search").Inc()
}
}()
go func() {
defer wg.Done()
mbResult, mbErr = s.provider.SearchArtists(ctx, query, limit, offset)
if mbErr == nil && len(mbResult.Items) > 0 {
if saveErr := s.artists.SaveAll(ctx, mbResult.Items); saveErr != nil {
log.Warn().Err(saveErr).Msg("failed to cache provider search results")
} else {
log.Debug().Int("count", len(mbResult.Items)).Msg("cached provider search results")
}
}
}()
wg.Wait()
if dbErr != nil {
log.Warn().Err(dbErr).Msg("database search failed")
}
if mbErr != nil {
log.Warn().Err(mbErr).Msg("provider search failed")
}
metrics.CacheMisses.WithLabelValues("artist_search").Inc()
log.Debug().Str("query", query).Str("provider", s.provider.Name()).Msg("artist search cache miss, querying provider")
return s.provider.SearchArtists(ctx, query, limit, offset)
if dbErr != nil && mbErr != nil {
return nil, mbErr
}
combined := s.deduplicateArtists(dbResult, mbResult)
log.Debug().
Int("db_count", countItems(dbResult)).
Int("provider_count", countItems(mbResult)).
Int("combined_count", len(combined.Items)).
Msg("artist search completed")
return combined, nil
}
func (s *MetadataService) deduplicateArtists(dbResult, mbResult *domain.SearchResult[domain.Artist]) *domain.SearchResult[domain.Artist] {
seen := make(map[string]bool)
var items []domain.Artist
total := 0
addArtist := func(artist domain.Artist) {
if artist.Type == "Other" {
return
}
for _, ext := range artist.ExternalIDs {
key := ext.Source + ":" + ext.SourceID
if seen[key] {
return
}
}
for _, ext := range artist.ExternalIDs {
seen[ext.Source+":"+ext.SourceID] = true
}
items = append(items, artist)
}
if dbResult != nil {
for _, artist := range dbResult.Items {
addArtist(artist)
}
total = dbResult.Total
}
if mbResult != nil {
for _, artist := range mbResult.Items {
addArtist(artist)
}
if mbResult.Total > total {
total = mbResult.Total
}
}
limit := 25
offset := 0
if dbResult != nil {
limit = dbResult.Limit
offset = dbResult.Offset
} else if mbResult != nil {
limit = mbResult.Limit
offset = mbResult.Offset
}
return &domain.SearchResult[domain.Artist]{
Items: items,
Total: total,
Limit: limit,
Offset: offset,
}
}
func countItems[T any](result *domain.SearchResult[T]) int {
if result == nil {
return 0
}
return len(result.Items)
}
func (s *MetadataService) SearchAlbums(ctx context.Context, query string, artist string, limit, offset int) (*domain.SearchResult[domain.Album], error) {