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", "exactness", "proximity", "attribute", "sort", "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": TranslitVariants(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": TranslitVariants(title), "artist_translit": TranslitVariants(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, TranslitVariants(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": TranslitVariants(title), "artist_translit": translit, }, nil }, }, } const batchSize = 1000 // Indexer полностью пересобирает индексы из PostgreSQL. Новые индексы собираются рядом // (`_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 }