memby/server/internal/api/search_stream.go

package api

import (
	"context"
	"encoding/json"
	"net/http"
	"net/url"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/ponzischeme89/memby/server/internal/store"
)

// Search updates are newline-delimited JSON rather than one large JSON document. The first
// update is deliberately allowed to come from whichever upstream answers first.
func (s *Server) handleSearchStream(w http.ResponseWriter, r *http.Request, sess store.Session) {
	term := strings.TrimSpace(r.URL.Query().Get("q"))
	if len([]rune(term)) < minSearchQueryRunes {
		writeJSON(w, http.StatusOK, map[string]any{"items": []json.RawMessage{}})
		return
	}
	limit := queryInt(r, "limit", 40, 100)
	s.recordSearchQuery(r.Context(), sess, term)
	canRequest := s.requestAllowed(r, sess)
	ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
	defer cancel()

	w.Header().Set("Content-Type", "application/x-ndjson")
	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("X-Accel-Buffering", "no")
	flusher, _ := w.(http.Flusher)
	enc := json.NewEncoder(w)
	updates := make(chan []json.RawMessage, 3)
	var wg sync.WaitGroup
	start := func(fn func(context.Context) []json.RawMessage) {
		wg.Add(1)
		go func() { defer wg.Done(); updates <- fn(ctx) }()
	}

	// Emby, Sonarr and Radarr are independent. In particular, an unavailable *arr must
	// never hold back the library answer.
	start(func(ctx context.Context) []json.RawMessage {
		result, err := s.emby.Items(ctx, credentials(sess), rowParams(url.Values{
			"SearchTerm": {term}, "IncludeItemTypes": {"Movie,Series,Episode"},
			"Recursive": {"true"}, "Limit": {itoa(limit)},
		}, fieldsRow+",ProviderIds"))
		if err != nil {
			return nil
		}
		items := s.personalizeSearch(ctx, sess, term, result.Items, limit)
		s.decorateItems(ctx, items)
		return markSearchAvailable(items)
	})
	if s.sonarrEnabled(ctx) {
		start(func(ctx context.Context) []json.RawMessage { return s.streamSonarr(ctx, term, limit, canRequest) })
	}
	if s.radarrEnabled(ctx) {
		start(func(ctx context.Context) []json.RawMessage { return s.streamRadarr(ctx, term, limit, canRequest) })
	}
	go func() { wg.Wait(); close(updates) }()

	merged := make([]json.RawMessage, 0, limit)
	for update := range updates {
		for _, item := range update {
			merged = mergeSearchRaw(merged, item, limit)
		}
		if err := enc.Encode(map[string]any{"items": merged}); err != nil {
			return
		}
		if flusher != nil {
			flusher.Flush()
		}
	}
}

func markSearchAvailable(items []json.RawMessage) []json.RawMessage {
	for i, raw := range items {
		var item map[string]any
		if json.Unmarshal(raw, &item) != nil {
			continue
		}
		item["MembySearchState"] = RequestStatusAvailable
		if encoded, err := json.Marshal(item); err == nil {
			items[i] = encoded
		}
	}
	return items
}

func (s *Server) streamSonarr(ctx context.Context, term string, limit int, canRequest bool) []json.RawMessage {
	series, err := s.sonarr.Lookup(ctx, term)
	if err != nil {
		return nil
	}
	ids := make([]int, 0, len(series))
	for _, v := range series {
		if v.TVDBID > 0 {
			ids = append(ids, v.TVDBID)
		}
	}
	inLibrary, _ := s.store.LibraryContainsProviderIDs(ctx, "Tvdb", ids)
	items := make([]json.RawMessage, 0, limit)
	for _, v := range series {
		if v.TVDBID == 0 || len(items) >= limit {
			continue
		}
		state := lookupStatusFor(RequestSubject{Tracked: v.ID > 0, InLibrary: inLibrary[v.TVDBID], Released: seriesReleased(v.Status, v.NextAiring, time.Now())}, false)
		items = append(items, searchExternalItem(v.Title, v.Year, v.Overview, "Series", "sonarr", state,
			strconv.Itoa(v.TVDBID), sonarrCoverURL(v.Images, "poster"), canRequest))
	}
	return items
}

func (s *Server) streamRadarr(ctx context.Context, term string, limit int, canRequest bool) []json.RawMessage {
	movies, err := s.radarr.Lookup(ctx, term)
	if err != nil {
		return nil
	}
	ids := make([]int, 0, len(movies))
	for _, v := range movies {
		if v.TMDBID > 0 {
			ids = append(ids, v.TMDBID)
		}
	}
	inLibrary, _ := s.store.LibraryContainsProviderIDs(ctx, "Tmdb", ids)
	items := make([]json.RawMessage, 0, limit)
	for _, v := range movies {
		if v.TMDBID == 0 || len(items) >= limit {
			continue
		}
		state := lookupStatusFor(RequestSubject{Tracked: v.ID > 0, HasFile: v.HasFile, InLibrary: inLibrary[v.TMDBID], Released: movieReleased(v.Status)}, false)
		items = append(items, searchExternalItem(v.Title, v.Year, v.Overview, "Movie", "radarr", state,
			strconv.Itoa(v.TMDBID), radarrCoverURL(v.Images, "poster"), canRequest))
	}
	return items
}

func searchExternalItem(title string, year int, overview, itemType, source, state, providerID, poster string, canRequest bool) json.RawMessage {
	item := map[string]any{"Id": source + ":" + providerID, "Name": title, "Type": itemType,
		"ProductionYear": year, "Overview": overview, "MembySource": source,
		"MembySearchState": state, "MembyPlayable": false, "MembyPosterURL": poster,
		"MembyRequestable": canRequest && state == RequestStatusRequestable,
		"ProviderIds":      map[string]string{map[string]string{"sonarr": "Tvdb", "radarr": "Tmdb"}[source]: providerID}}
	raw, _ := json.Marshal(item)
	return raw
}

func mergeSearchRaw(existing []json.RawMessage, candidate json.RawMessage, limit int) []json.RawMessage {
	key := searchRawKey(candidate)
	for i, current := range existing {
		if searchRawKey(current) != key {
			continue
		}
		var currentItem map[string]any
		_ = json.Unmarshal(current, &currentItem)
		// An Emby item is authoritative and stays playable even when an *arr later finds it.
		if source, _ := currentItem["MembySource"].(string); source == "" {
			var incoming map[string]any
			_ = json.Unmarshal(candidate, &incoming)
			if incoming["MembySource"] != nil {
				return existing
			}
			return existing
		}
		var merged map[string]any
		_ = json.Unmarshal(current, &merged)
		var incoming map[string]any
		_ = json.Unmarshal(candidate, &incoming)
		if incoming["MembySource"] == nil {
			existing[i] = candidate
			return existing
		}
		for k, v := range incoming {
			merged[k] = v
		}
		existing[i], _ = json.Marshal(merged)
		return existing
	}
	if len(existing) < limit {
		existing = append(existing, candidate)
	}
	return existing
}

func searchRawKey(raw json.RawMessage) string {
	var item struct {
		ProviderIDs map[string]string `json:"ProviderIds"`
		Name, Type  string
	}
	if json.Unmarshal(raw, &item) == nil {
		for _, key := range []string{"Tmdb", "Tvdb", "Imdb"} {
			if id := item.ProviderIDs[key]; id != "" {
				return key + ":" + id
			}
		}
	}
	return strings.ToLower(item.Type + ":" + strings.TrimSpace(item.Name))
}