243 lines
7.1 KiB
Go
243 lines
7.1 KiB
Go
package search
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"log/slog"
|
|
"net/http"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
// Индексы Meilisearch.
|
|
const (
|
|
IndexArtists = "artists"
|
|
IndexAlbums = "albums"
|
|
IndexTracks = "tracks"
|
|
)
|
|
|
|
type indexSpec struct {
|
|
uid string
|
|
settings map[string]any
|
|
query string
|
|
doc func(pgx.Rows) (map[string]any, error)
|
|
}
|
|
|
|
var rankingRules = []string{"words", "typo", "proximity", "attribute", "sort", "exactness", "popularity:desc"}
|
|
|
|
var specs = []indexSpec{
|
|
{
|
|
uid: IndexArtists,
|
|
settings: map[string]any{
|
|
"searchableAttributes": []string{"name", "name_translit"},
|
|
"sortableAttributes": []string{"popularity"},
|
|
"rankingRules": rankingRules,
|
|
},
|
|
query: `SELECT a.id::text, a.name FROM artists a`,
|
|
doc: func(r pgx.Rows) (map[string]any, error) {
|
|
var id, name string
|
|
if err := r.Scan(&id, &name); err != nil {
|
|
return nil, err
|
|
}
|
|
return map[string]any{"id": id, "name": name, "name_translit": Translit(name), "popularity": 0}, nil
|
|
},
|
|
},
|
|
{
|
|
uid: IndexAlbums,
|
|
settings: map[string]any{
|
|
"searchableAttributes": []string{"title", "artist_name", "title_translit", "artist_translit"},
|
|
"filterableAttributes": []string{"year"},
|
|
"sortableAttributes": []string{"year", "popularity"},
|
|
"rankingRules": rankingRules,
|
|
},
|
|
query: `SELECT al.id::text, al.title, aa.name, al.year FROM albums al JOIN artists aa ON aa.id = al.album_artist_id`,
|
|
doc: func(r pgx.Rows) (map[string]any, error) {
|
|
var (
|
|
id, title, artist string
|
|
year *int
|
|
)
|
|
if err := r.Scan(&id, &title, &artist, &year); err != nil {
|
|
return nil, err
|
|
}
|
|
return map[string]any{
|
|
"id": id, "title": title, "artist_name": artist, "year": year, "popularity": 0,
|
|
"title_translit": Translit(title), "artist_translit": Translit(artist),
|
|
}, nil
|
|
},
|
|
},
|
|
{
|
|
uid: IndexTracks,
|
|
settings: map[string]any{
|
|
"searchableAttributes": []string{"title", "artist_names", "album_title", "title_translit", "artist_translit"},
|
|
"filterableAttributes": []string{"genres", "year", "available"},
|
|
"sortableAttributes": []string{"year", "popularity"},
|
|
"rankingRules": rankingRules,
|
|
},
|
|
query: `
|
|
SELECT t.id::text, t.title, al.title, COALESCE(t.year, al.year),
|
|
COALESCE((SELECT array_agg(a.name ORDER BY ta.position) FROM track_artists ta
|
|
JOIN artists a ON a.id = ta.artist_id WHERE ta.track_id = t.id), '{}'),
|
|
COALESCE((SELECT array_agg(g.name) FROM track_genres tg JOIN genres g ON g.id = tg.genre_id
|
|
WHERE tg.track_id = t.id), '{}'),
|
|
t.primary_file_id IS NOT NULL
|
|
FROM tracks t JOIN albums al ON al.id = t.album_id`,
|
|
doc: func(r pgx.Rows) (map[string]any, error) {
|
|
var (
|
|
id, title, album string
|
|
year *int
|
|
artists, genres []string
|
|
available bool
|
|
)
|
|
if err := r.Scan(&id, &title, &album, &year, &artists, &genres, &available); err != nil {
|
|
return nil, err
|
|
}
|
|
translit := make([]string, 0, len(artists))
|
|
for _, a := range artists {
|
|
translit = append(translit, Translit(a))
|
|
}
|
|
return map[string]any{
|
|
"id": id, "title": title, "album_title": album, "year": year, "artist_names": artists,
|
|
"genres": genres, "available": available, "popularity": 0,
|
|
"title_translit": Translit(title), "artist_translit": translit,
|
|
}, nil
|
|
},
|
|
},
|
|
}
|
|
|
|
const batchSize = 1000
|
|
|
|
// Indexer полностью пересобирает индексы из PostgreSQL. Новые индексы собираются рядом
|
|
// (`<uid>_next`) и подменяются через swap-indexes — поиск работает всё время.
|
|
type Indexer struct {
|
|
pool *pgxpool.Pool
|
|
meili *Meili
|
|
log *slog.Logger
|
|
|
|
mu sync.Mutex
|
|
running bool
|
|
again bool
|
|
// LastError — ошибка последней переиндексации (для статуса), пусто — успех.
|
|
lastErr error
|
|
lastDone time.Time
|
|
}
|
|
|
|
// NewIndexer создаёт индексатор.
|
|
func NewIndexer(pool *pgxpool.Pool, meili *Meili, log *slog.Logger) *Indexer {
|
|
return &Indexer{pool: pool, meili: meili, log: log}
|
|
}
|
|
|
|
// Trigger запускает переиндексацию в фоне. Если она уже идёт — ещё одна выполнится сразу после.
|
|
func (ix *Indexer) Trigger() {
|
|
ix.mu.Lock()
|
|
if ix.running {
|
|
ix.again = true
|
|
ix.mu.Unlock()
|
|
return
|
|
}
|
|
ix.running = true
|
|
ix.mu.Unlock()
|
|
go func() {
|
|
for {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute)
|
|
err := ix.Reindex(ctx)
|
|
cancel()
|
|
if err != nil {
|
|
ix.log.Error("search reindex failed", "err", err)
|
|
}
|
|
ix.mu.Lock()
|
|
ix.lastErr, ix.lastDone = err, time.Now()
|
|
if !ix.again {
|
|
ix.running = false
|
|
ix.mu.Unlock()
|
|
return
|
|
}
|
|
ix.again = false
|
|
ix.mu.Unlock()
|
|
}
|
|
}()
|
|
}
|
|
|
|
// Reindex пересобирает все индексы.
|
|
func (ix *Indexer) Reindex(ctx context.Context) error {
|
|
started := time.Now()
|
|
for _, s := range specs {
|
|
if err := ix.rebuild(ctx, s); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
ix.log.Info("search reindexed", "duration_ms", time.Since(started).Milliseconds())
|
|
return nil
|
|
}
|
|
|
|
func (ix *Indexer) rebuild(ctx context.Context, s indexSpec) error {
|
|
next := s.uid + "_next"
|
|
if err := ix.meili.deleteIndexIfExists(ctx, next); err != nil {
|
|
return err
|
|
}
|
|
if err := ix.meili.call(ctx, http.MethodPost, "/indexes", map[string]any{"uid": next, "primaryKey": "id"}); err != nil {
|
|
return err
|
|
}
|
|
if err := ix.meili.call(ctx, http.MethodPatch, "/indexes/"+next+"/settings", s.settings); err != nil {
|
|
return err
|
|
}
|
|
|
|
rows, err := ix.pool.Query(ctx, s.query)
|
|
if err != nil {
|
|
return fmt.Errorf("search: %s: %w", s.uid, err)
|
|
}
|
|
batch := make([]map[string]any, 0, batchSize)
|
|
flush := func() error {
|
|
if len(batch) == 0 {
|
|
return nil
|
|
}
|
|
err := ix.meili.call(ctx, http.MethodPost, "/indexes/"+next+"/documents", batch)
|
|
batch = batch[:0]
|
|
return err
|
|
}
|
|
for rows.Next() {
|
|
doc, err := s.doc(rows)
|
|
if err != nil {
|
|
rows.Close()
|
|
return fmt.Errorf("search: %s: %w", s.uid, err)
|
|
}
|
|
batch = append(batch, doc)
|
|
if len(batch) == batchSize {
|
|
if err := flush(); err != nil {
|
|
rows.Close()
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
rows.Close()
|
|
if err := rows.Err(); err != nil {
|
|
return fmt.Errorf("search: %s: %w", s.uid, err)
|
|
}
|
|
if err := flush(); err != nil {
|
|
return err
|
|
}
|
|
|
|
// swap-indexes требует, чтобы оба индекса существовали.
|
|
if err := ix.meili.do(ctx, http.MethodGet, "/indexes/"+s.uid, nil, nil); errors.Is(err, ErrMeiliNotFound) {
|
|
if err := ix.meili.call(ctx, http.MethodPost, "/indexes", map[string]any{"uid": s.uid, "primaryKey": "id"}); err != nil {
|
|
return err
|
|
}
|
|
} else if err != nil {
|
|
return err
|
|
}
|
|
if err := ix.meili.call(ctx, http.MethodPost, "/swap-indexes", []map[string]any{{"indexes": []string{s.uid, next}}}); err != nil {
|
|
return err
|
|
}
|
|
return ix.meili.call(ctx, http.MethodDelete, "/indexes/"+next, nil)
|
|
}
|
|
|
|
// State — состояние индексатора для админки.
|
|
func (ix *Indexer) State() (running bool, lastDone time.Time, lastErr error) {
|
|
ix.mu.Lock()
|
|
defer ix.mu.Unlock()
|
|
return ix.running, ix.lastDone, ix.lastErr
|
|
}
|