Alexander
b4e763be44
Replace per-album loadRelations calls with batch queries using ANY($1). Reduces 1+2N queries to 3 queries total (1.45s → 50ms for 83 albums).
452 lines
12 KiB
Go
452 lines
12 KiB
Go
package postgres
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
|
|
"github.com/metadata-agregator/internal/domain"
|
|
"github.com/metadata-agregator/internal/repository"
|
|
)
|
|
|
|
type AlbumRepository struct {
|
|
pool *pgxpool.Pool
|
|
}
|
|
|
|
func NewAlbumRepository(pool *pgxpool.Pool) *AlbumRepository {
|
|
return &AlbumRepository{pool: pool}
|
|
}
|
|
|
|
func (r *AlbumRepository) GetByID(ctx context.Context, id string) (*domain.Album, error) {
|
|
query := `
|
|
SELECT id, title, album_type, secondary_types, release_date, upc, total_tracks,
|
|
total_discs, cover_url, source, source_id
|
|
FROM albums
|
|
WHERE id = $1`
|
|
|
|
album, err := r.scanAlbum(ctx, query, id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if err := r.loadRelations(ctx, album); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return album, nil
|
|
}
|
|
|
|
func (r *AlbumRepository) GetByExternalID(ctx context.Context, source, sourceID string) (*domain.Album, error) {
|
|
query := `
|
|
SELECT a.id, a.title, a.album_type, a.secondary_types, a.release_date, a.upc,
|
|
a.total_tracks, a.total_discs, a.cover_url, a.source, a.source_id
|
|
FROM albums a
|
|
JOIN album_external_ids e ON a.id = e.album_id
|
|
WHERE e.source = $1 AND e.source_id = $2`
|
|
|
|
album, err := r.scanAlbum(ctx, query, source, sourceID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if err := r.loadRelations(ctx, album); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return album, nil
|
|
}
|
|
|
|
func (r *AlbumRepository) GetByArtistID(ctx context.Context, artistID string, limit, offset int) (*domain.SearchResult[domain.Album], error) {
|
|
countQuery := `
|
|
SELECT COUNT(DISTINCT a.id)
|
|
FROM albums a
|
|
JOIN album_artists aa ON a.id = aa.album_id
|
|
JOIN artist_external_ids ae ON aa.artist_id = ae.artist_id
|
|
WHERE ae.source_id = $1`
|
|
|
|
searchQuery := `
|
|
SELECT DISTINCT a.id, a.title, a.album_type, a.secondary_types, a.release_date,
|
|
a.upc, a.total_tracks, a.total_discs, a.cover_url, a.source, a.source_id
|
|
FROM albums a
|
|
JOIN album_artists aa ON a.id = aa.album_id
|
|
JOIN artist_external_ids ae ON aa.artist_id = ae.artist_id
|
|
WHERE ae.source_id = $1
|
|
ORDER BY a.release_date DESC NULLS LAST
|
|
LIMIT $2 OFFSET $3`
|
|
|
|
var total int
|
|
if err := r.pool.QueryRow(ctx, countQuery, artistID).Scan(&total); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
rows, err := r.pool.Query(ctx, searchQuery, artistID, limit, offset)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var albums []domain.Album
|
|
for rows.Next() {
|
|
album, err := r.scanAlbumFromRow(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
albums = append(albums, *album)
|
|
}
|
|
|
|
return &domain.SearchResult[domain.Album]{
|
|
Items: albums,
|
|
Total: total,
|
|
Limit: limit,
|
|
Offset: offset,
|
|
}, nil
|
|
}
|
|
|
|
func (r *AlbumRepository) GetAllByArtistID(ctx context.Context, artistID string) ([]domain.Album, error) {
|
|
query := `
|
|
SELECT DISTINCT a.id, a.title, a.album_type, a.secondary_types, a.release_date,
|
|
a.upc, a.total_tracks, a.total_discs, a.cover_url, a.source, a.source_id
|
|
FROM albums a
|
|
JOIN album_artists aa ON a.id = aa.album_id
|
|
JOIN artist_external_ids ae ON aa.artist_id = ae.artist_id
|
|
WHERE ae.source_id = $1
|
|
ORDER BY a.release_date DESC NULLS LAST`
|
|
|
|
rows, err := r.pool.Query(ctx, query, artistID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var albums []domain.Album
|
|
var albumIDs []string
|
|
albumIndex := make(map[string]int)
|
|
|
|
for rows.Next() {
|
|
album, err := r.scanAlbumFromRow(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
albumIndex[album.ID] = len(albums)
|
|
albumIDs = append(albumIDs, album.ID)
|
|
albums = append(albums, *album)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if len(albums) == 0 {
|
|
return albums, nil
|
|
}
|
|
|
|
if err := r.loadRelationsBatch(ctx, albums, albumIDs, albumIndex); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return albums, nil
|
|
}
|
|
|
|
func (r *AlbumRepository) loadRelationsBatch(ctx context.Context, albums []domain.Album, albumIDs []string, albumIndex map[string]int) error {
|
|
extQuery := `SELECT album_id, source, source_id, url FROM album_external_ids WHERE album_id = ANY($1)`
|
|
extRows, err := r.pool.Query(ctx, extQuery, albumIDs)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer extRows.Close()
|
|
|
|
for extRows.Next() {
|
|
var albumID string
|
|
var ext domain.ExternalID
|
|
var url *string
|
|
if err := extRows.Scan(&albumID, &ext.Source, &ext.SourceID, &url); err != nil {
|
|
return err
|
|
}
|
|
ext.URL = derefString(url)
|
|
if idx, ok := albumIndex[albumID]; ok {
|
|
albums[idx].ExternalIDs = append(albums[idx].ExternalIDs, ext)
|
|
}
|
|
}
|
|
if err := extRows.Err(); err != nil {
|
|
return err
|
|
}
|
|
|
|
artistQuery := `
|
|
SELECT aa.album_id, a.id, a.name, a.sort_name, a.artist_type, a.country,
|
|
aa.role, aa.position
|
|
FROM album_artists aa
|
|
JOIN artists a ON a.id = aa.artist_id
|
|
WHERE aa.album_id = ANY($1)
|
|
ORDER BY aa.album_id, aa.position`
|
|
|
|
artistRows, err := r.pool.Query(ctx, artistQuery, albumIDs)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer artistRows.Close()
|
|
|
|
for artistRows.Next() {
|
|
var (
|
|
albumID string
|
|
ac domain.ArtistCredit
|
|
sortName *string
|
|
artType *string
|
|
country *string
|
|
)
|
|
if err := artistRows.Scan(
|
|
&albumID, &ac.Artist.ID, &ac.Artist.Name, &sortName, &artType, &country,
|
|
&ac.Role, &ac.Position,
|
|
); err != nil {
|
|
return err
|
|
}
|
|
ac.Artist.SortName = derefString(sortName)
|
|
ac.Artist.Type = derefString(artType)
|
|
ac.Artist.Country = derefString(country)
|
|
if idx, ok := albumIndex[albumID]; ok {
|
|
albums[idx].Artists = append(albums[idx].Artists, ac)
|
|
}
|
|
}
|
|
|
|
return artistRows.Err()
|
|
}
|
|
|
|
func (r *AlbumRepository) SaveAll(ctx context.Context, albums []domain.Album) error {
|
|
tx, err := r.pool.Begin(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
|
|
for i := range albums {
|
|
album := &albums[i]
|
|
|
|
var source, sourceID string
|
|
if len(album.ExternalIDs) > 0 {
|
|
source = album.ExternalIDs[0].Source
|
|
sourceID = album.ExternalIDs[0].SourceID
|
|
}
|
|
|
|
query := `
|
|
INSERT INTO albums (id, title, album_type, secondary_types, release_date, upc,
|
|
total_tracks, total_discs, cover_url, source, source_id)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
|
|
ON CONFLICT (id) DO UPDATE SET
|
|
title = EXCLUDED.title,
|
|
album_type = EXCLUDED.album_type,
|
|
secondary_types = EXCLUDED.secondary_types,
|
|
release_date = EXCLUDED.release_date,
|
|
upc = EXCLUDED.upc,
|
|
total_tracks = EXCLUDED.total_tracks,
|
|
total_discs = EXCLUDED.total_discs,
|
|
cover_url = EXCLUDED.cover_url,
|
|
updated_at = now()`
|
|
|
|
_, err = tx.Exec(ctx, query,
|
|
album.ID, album.Title, nullString(album.Type), album.SecondaryTypes,
|
|
album.ReleaseDate, nullString(album.UPC), album.TotalTracks, album.TotalDiscs,
|
|
nullString(album.CoverURL), source, sourceID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
for _, ext := range album.ExternalIDs {
|
|
extQuery := `
|
|
INSERT INTO album_external_ids (album_id, source, source_id, url)
|
|
VALUES ($1, $2, $3, $4)
|
|
ON CONFLICT (album_id, source, source_id) DO UPDATE SET
|
|
url = EXCLUDED.url,
|
|
fetched_at = now()`
|
|
|
|
_, err = tx.Exec(ctx, extQuery, album.ID, ext.Source, ext.SourceID, nullString(ext.URL))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
for _, ac := range album.Artists {
|
|
artistQuery := `
|
|
INSERT INTO album_artists (album_id, artist_id, role, position)
|
|
VALUES ($1, $2, $3, $4)
|
|
ON CONFLICT (album_id, artist_id, role) DO NOTHING`
|
|
|
|
_, err = tx.Exec(ctx, artistQuery, album.ID, ac.Artist.ID, ac.Role, ac.Position)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
|
|
return tx.Commit(ctx)
|
|
}
|
|
|
|
func (r *AlbumRepository) Save(ctx context.Context, album *domain.Album) error {
|
|
tx, err := r.pool.Begin(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
|
|
var source, sourceID string
|
|
if len(album.ExternalIDs) > 0 {
|
|
source = album.ExternalIDs[0].Source
|
|
sourceID = album.ExternalIDs[0].SourceID
|
|
}
|
|
|
|
query := `
|
|
INSERT INTO albums (id, title, album_type, secondary_types, release_date, upc,
|
|
total_tracks, total_discs, cover_url, source, source_id)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
|
|
ON CONFLICT (id) DO UPDATE SET
|
|
title = EXCLUDED.title,
|
|
album_type = EXCLUDED.album_type,
|
|
secondary_types = EXCLUDED.secondary_types,
|
|
release_date = EXCLUDED.release_date,
|
|
upc = EXCLUDED.upc,
|
|
total_tracks = EXCLUDED.total_tracks,
|
|
total_discs = EXCLUDED.total_discs,
|
|
cover_url = EXCLUDED.cover_url,
|
|
updated_at = now()`
|
|
|
|
_, err = tx.Exec(ctx, query,
|
|
album.ID, album.Title, nullString(album.Type), album.SecondaryTypes,
|
|
album.ReleaseDate, nullString(album.UPC), album.TotalTracks, album.TotalDiscs,
|
|
nullString(album.CoverURL), source, sourceID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
for _, ext := range album.ExternalIDs {
|
|
extQuery := `
|
|
INSERT INTO album_external_ids (album_id, source, source_id, url)
|
|
VALUES ($1, $2, $3, $4)
|
|
ON CONFLICT (album_id, source, source_id) DO UPDATE SET
|
|
url = EXCLUDED.url,
|
|
fetched_at = now()`
|
|
|
|
_, err = tx.Exec(ctx, extQuery, album.ID, ext.Source, ext.SourceID, nullString(ext.URL))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
for _, ac := range album.Artists {
|
|
artistQuery := `
|
|
INSERT INTO album_artists (album_id, artist_id, role, position)
|
|
VALUES ($1, $2, $3, $4)
|
|
ON CONFLICT (album_id, artist_id, role) DO NOTHING`
|
|
|
|
_, err = tx.Exec(ctx, artistQuery, album.ID, ac.Artist.ID, ac.Role, ac.Position)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return tx.Commit(ctx)
|
|
}
|
|
|
|
func (r *AlbumRepository) scanAlbum(ctx context.Context, query string, args ...any) (*domain.Album, error) {
|
|
row := r.pool.QueryRow(ctx, query, args...)
|
|
return r.scanAlbumRow(row)
|
|
}
|
|
|
|
func (r *AlbumRepository) scanAlbumFromRow(row pgx.Row) (*domain.Album, error) {
|
|
return r.scanAlbumRow(row)
|
|
}
|
|
|
|
func (r *AlbumRepository) scanAlbumRow(row pgx.Row) (*domain.Album, error) {
|
|
var (
|
|
album domain.Album
|
|
albumType *string
|
|
releaseDate *time.Time
|
|
upc *string
|
|
totalTracks *int
|
|
totalDiscs *int
|
|
coverURL *string
|
|
source string
|
|
sourceID *string
|
|
)
|
|
|
|
err := row.Scan(
|
|
&album.ID, &album.Title, &albumType, &album.SecondaryTypes, &releaseDate, &upc,
|
|
&totalTracks, &totalDiscs, &coverURL, &source, &sourceID,
|
|
)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, repository.ErrNotFound
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
album.Type = derefString(albumType)
|
|
album.ReleaseDate = releaseDate
|
|
album.UPC = derefString(upc)
|
|
if totalTracks != nil {
|
|
album.TotalTracks = *totalTracks
|
|
}
|
|
if totalDiscs != nil {
|
|
album.TotalDiscs = *totalDiscs
|
|
}
|
|
album.CoverURL = derefString(coverURL)
|
|
|
|
return &album, nil
|
|
}
|
|
|
|
func (r *AlbumRepository) loadRelations(ctx context.Context, album *domain.Album) error {
|
|
extQuery := `SELECT source, source_id, url FROM album_external_ids WHERE album_id = $1`
|
|
extRows, err := r.pool.Query(ctx, extQuery, album.ID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer extRows.Close()
|
|
|
|
for extRows.Next() {
|
|
var ext domain.ExternalID
|
|
var url *string
|
|
if err := extRows.Scan(&ext.Source, &ext.SourceID, &url); err != nil {
|
|
return err
|
|
}
|
|
ext.URL = derefString(url)
|
|
album.ExternalIDs = append(album.ExternalIDs, ext)
|
|
}
|
|
if err := extRows.Err(); err != nil {
|
|
return err
|
|
}
|
|
|
|
artistQuery := `
|
|
SELECT a.id, a.name, a.sort_name, a.artist_type, a.country,
|
|
aa.role, aa.position
|
|
FROM album_artists aa
|
|
JOIN artists a ON a.id = aa.artist_id
|
|
WHERE aa.album_id = $1
|
|
ORDER BY aa.position`
|
|
|
|
artistRows, err := r.pool.Query(ctx, artistQuery, album.ID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer artistRows.Close()
|
|
|
|
for artistRows.Next() {
|
|
var (
|
|
ac domain.ArtistCredit
|
|
sortName *string
|
|
artType *string
|
|
country *string
|
|
)
|
|
if err := artistRows.Scan(
|
|
&ac.Artist.ID, &ac.Artist.Name, &sortName, &artType, &country,
|
|
&ac.Role, &ac.Position,
|
|
); err != nil {
|
|
return err
|
|
}
|
|
ac.Artist.SortName = derefString(sortName)
|
|
ac.Artist.Type = derefString(artType)
|
|
ac.Artist.Country = derefString(country)
|
|
album.Artists = append(album.Artists, ac)
|
|
}
|
|
|
|
return artistRows.Err()
|
|
}
|