Files
silo-server/internal/api/handlers/catalog.go
203a18ae83 feat(observability): OpenTelemetry logs+traces with secret redaction and slog standardization (#290)
* feat(observability): OpenTelemetry logs+traces with secret redaction

Part of #265. Adds opt-in OpenTelemetry (logs + traces) alongside the existing
stderr + opslog pipeline, plus secret redaction on all sinks. Default-off: with
no OTEL_* / SILO_OTEL_ENABLED config, behavior is unchanged.

Bootstrap (internal/telemetry):
- Setup() builds one shared resource, a TracerProvider (parent-based trace-id
  ratio sampler), a LoggerProvider, and the W3C TraceContext+Baggage propagator
  from env. It installs NO MeterProvider — metrics stay on Prometheus, and the
  built-in no-op global MeterProvider keeps the trace instrumentation libs from
  double-emitting. Shutdown is deferred with a flush timeout.
- Logs are bridged via otelslog fan-out (slog.MultiHandler), level-gated by the
  shared LevelVar and best-effort so a failing collector can't break the console
  or DB branches. stderr + opslog stay untouched.

Secret redaction (internal/logredact):
- A slog.Handler masks secret-keyed attributes (password, token, api_key,
  authorization, cookie, ...) — including .With-bound attrs, nested groups,
  secret-keyed group subtrees, and values behind a LogValuer — on the console
  and OTLP sinks, with a no-op fast path when a record has no secret keys.
  opslog.shouldRedact delegates to logredact.SecretKey so all sinks share one
  marker list.

Rotation is infra-managed (no custom file sink): container runtime for stderr,
collector/backend for OTLP, opslog partition-pruning for the DB. Documented in
docs/architecture/observability.md.

Verification: go build ./..., go vet, gofmt -l — clean; go test
./internal/telemetry/ ./internal/logredact/ -race pass.

AI-use disclosure: implemented with AI assistance (Claude Code), including
adversarial reviews that hardened the bootstrap and fixed two redaction leak
paths; reviewed by the author.

* refactor(observability): slog context+component sweep, sloglint gate (phase 3)

Part of #265. Builds on the OTel bootstrap + redaction commit.

Standardizes every log call site onto the context-carrying slog variants so
records correlate with the active OpenTelemetry trace, and locks the standard
in with a machine gate so future code (human- or AI-authored) can't drift back.

- Call-site sweep: converted the remaining slog.<Level>(...) calls to the
  slog.<Level>Context(ctx, ...) form wherever a context.Context is in scope
  (background/init calls with no ctx are left as-is), across 183 files. Applied
  via a type-aware AST codemod. Log levels and message strings are preserved
  verbatim; a component attr (canonical per-package name) is added to direct
  package-level slog calls. Bound-logger calls keep their existing .With
  bindings. The main.go and telemetry package conversions rode with their file
  in the previous commit to keep each file within a single commit.
- Enforcement (.golangci.yml): enable sloglint with context=scope, static-msg,
  key-naming-case=snake, no-mixed-args. After the sweep all four report zero
  violations repo-wide (tests included), so make lint / CI now blocks any
  regression to the non-context form. The gate ships with the sweep because it
  cannot be green until the legacy sites are converted.

Metrics remain on Prometheus; no behavior change to /metrics or Grafana.

Verification: go build ./..., go vet ./..., gofmt -l — clean; sloglint (all 4
rules) 0 violations repo-wide; log levels verified unchanged.

AI-use disclosure: implemented with AI assistance (Claude Code), including the
codemod; reviewed by the author.

* fix(observability): honor per-signal OTLP protocol and secret WithGroup names

Two Codex review findings on PR #290:

- telemetry: OTEL_EXPORTER_OTLP_{TRACES,LOGS}_PROTOCOL now override the
  generic OTEL_EXPORTER_OTLP_PROTOCOL per signal, so mixed collector
  setups (e.g. HTTP logs + gRPC traces) build the right exporter.
- logredact: entering a group whose name is secret-bearing (e.g.
  WithGroup("authorization")) now masks every leaf in that subtree,
  matching how slog.Group("authorization", ...) is masked as a whole.

* fix(observability): address review feedback on telemetry bootstrap

- Telemetry setup failure no longer kills boot: Setup returns usable
  no-op providers alongside the error and main logs and continues with
  telemetry disabled, honoring the best-effort contract.
- Honor OTEL_TRACES_SAMPLER (always_on/off, traceidratio, parentbased_*
  variants); unsupported values fall back to parentbased_traceidratio.
- Attach node identity as semconv service.instance.id instead of the
  non-semconv node.name.
- Rename opslog retention-scope log attrs to target_component/target_level
  so they no longer collide with the canonical component routing key, and
  tag those lines with component=opslog.
- Fix stale levelGated comment casing; use WarnContext in the telemetry
  shutdown defer; document the LogValuer double-resolve on the redaction
  slow path.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

---------

Co-authored-by: Quick <31828688+Quick104@users.noreply.github.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
2026-07-09 08:53:52 -04:00

751 lines
24 KiB
Go

package handlers
import (
"encoding/json"
"errors"
"fmt"
"log/slog"
"net/http"
"strconv"
"strings"
"sync"
"time"
apimw "github.com/Silo-Server/silo-server/internal/api/middleware"
"github.com/Silo-Server/silo-server/internal/catalog"
"github.com/Silo-Server/silo-server/internal/models"
"github.com/Silo-Server/silo-server/internal/sections"
)
// audiobookGroupsCacheTTL matches the client's React Query staleTime: the
// grouped author/narrator browse is cached server-side for the same window so
// the sequential page fetches and a quick refresh reuse one aggregation.
const audiobookGroupsCacheTTL = 60 * time.Second
type CatalogHandler struct {
resolver *catalog.CatalogResolver
itemsH *ItemsHandler
workSummary catalog.WorkSummaryProvider
groupsCacheOnce sync.Once
groupsCache *catalog.AudiobookGroupsCache
}
// audiobookGroups returns the lazily-initialized grouped-browse cache. Built on
// first use (only the route-mounted singleton handler serves audiobook-groups)
// so the per-request CatalogHandler instances never spawn a cache sweeper.
func (h *CatalogHandler) audiobookGroups() *catalog.AudiobookGroupsCache {
h.groupsCacheOnce.Do(func() {
h.groupsCache = catalog.NewAudiobookGroupsCache(h.itemsH.browseRepo.Pool(), audiobookGroupsCacheTTL)
})
return h.groupsCache
}
func NewCatalogHandler(resolver *catalog.CatalogResolver, itemsH *ItemsHandler) *CatalogHandler {
return &CatalogHandler{
resolver: resolver,
itemsH: itemsH,
}
}
func (h *CatalogHandler) SetWorkSummaryProvider(provider catalog.WorkSummaryProvider) {
h.workSummary = provider
}
type catalogResponse struct {
Total int `json:"total"`
TotalExact bool `json:"total_exact"`
HasMore bool `json:"has_more"`
Items []itemListResponse `json:"items"`
Snapshot string `json:"snapshot,omitempty"`
SearchDiagnostics *searchDiagnostics `json:"search_diagnostics,omitempty"`
}
// searchDiagnostics is an additive, per-query observability object emitted on
// /api/v1/catalog only when a relevance-sorted search actually ran through a
// CatalogSearchProvider. mode/semantic_used reflect POST-downgrade reality
// (a hybrid request that fell back to keyword reports mode="keyword",
// semantic_used=false). fallback_reason and index_pending_updates are omitted
// when empty.
type searchDiagnostics struct {
Provider string `json:"provider"`
Mode string `json:"mode"`
SemanticUsed bool `json:"semantic_used"`
FallbackReason string `json:"fallback_reason,omitempty"`
IndexPendingUpdates int `json:"index_pending_updates,omitempty"`
}
type catalogFiltersResponse struct {
Genres []string `json:"genres"`
Studios []string `json:"studios"`
Networks []string `json:"networks"`
Countries []string `json:"countries"`
OriginalLanguages []string `json:"original_languages"`
ContentRatings []string `json:"content_ratings"`
Authors []string `json:"authors"`
Narrators []string `json:"narrators"`
Series []string `json:"series"`
Resolutions *[]string `json:"resolutions,omitempty"`
AudioLanguages *[]string `json:"audio_languages,omitempty"`
SubtitleLanguages *[]string `json:"subtitle_languages,omitempty"`
}
func (h *CatalogHandler) HandleGetCatalog(w http.ResponseWriter, r *http.Request) {
if h == nil || h.resolver == nil || h.itemsH == nil {
writeError(w, http.StatusInternalServerError, "internal_error", "Catalog is not configured")
return
}
req, err := catalog.ParseCatalogRequest(r.URL.Query())
if err != nil {
writeError(w, http.StatusBadRequest, "bad_request", err.Error())
return
}
accessFilter := h.itemsH.accessFilter(r)
groupedByWork := strings.EqualFold(strings.TrimSpace(r.URL.Query().Get("group")), "work")
if groupedByWork {
result, entries, err := h.resolveGroupedCatalogByWork(r, req, accessFilter)
if err != nil {
if errors.Is(err, catalog.ErrInvalidCatalogRequest) {
writeError(w, http.StatusBadRequest, "bad_request", err.Error())
return
}
if errors.Is(err, catalog.ErrCatalogSourceNotFound) {
writeError(w, http.StatusNotFound, "not_found", "Catalog source not found")
return
}
slog.ErrorContext(r.Context(), "catalog: resolve grouped by work failed", "component", "api", "err_msg", err.Error())
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to resolve catalog")
return
}
resultItems := groupedCatalogItems(entries)
items := h.catalogItemResponses(r, resultItems, catalog.NormalizeQuerySort(req.Query.Sort).Field, accessFilter)
for i := range items {
if i < len(entries) && entries[i].summary != nil {
applyWorkSummaryToCatalogItem(&items[i], entries[i].summary)
}
}
h.writeCatalogResponse(w, result, items, groupedByWork)
return
}
result, err := h.resolver.Resolve(r.Context(), req, accessFilter)
if err != nil {
if errors.Is(err, catalog.ErrInvalidCatalogRequest) {
writeError(w, http.StatusBadRequest, "bad_request", err.Error())
return
}
if errors.Is(err, catalog.ErrCatalogSourceNotFound) {
writeError(w, http.StatusNotFound, "not_found", "Catalog source not found")
return
}
slog.ErrorContext(r.Context(), "catalog: resolve failed", "component", "api", "err_msg", err.Error())
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to resolve catalog")
return
}
items := h.catalogItemResponses(r, result.Items, catalog.NormalizeQuerySort(req.Query.Sort).Field, accessFilter)
h.writeCatalogResponse(w, result, items, groupedByWork)
}
func (h *CatalogHandler) catalogItemResponses(r *http.Request, resultItems []*models.MediaItem, sortField string, accessFilter catalog.AccessFilter) []itemListResponse {
var (
localizedItems []*models.MediaItem
overlaySummaries map[string]*models.OverlaySummary
userStates map[string]*itemUserStateResponse
episodeMetadata map[string]struct {
SeriesTitle string
SeasonNumber *int
EpisodeNumber *int
}
)
var enrichWG sync.WaitGroup
enrichWG.Add(4)
go func() {
defer enrichWG.Done()
localizedItems = h.itemsH.localizeItemListModels(r.Context(), resultItems, accessFilter)
}()
go func() {
defer enrichWG.Done()
overlaySummaries = h.itemsH.listOverlaySummaries(r.Context(), resultItems, accessFilter)
}()
go func() {
defer enrichWG.Done()
userStates = h.itemsH.listItemUserStates(r, resultItems)
}()
go func() {
defer enrichWG.Done()
episodeMetadata = h.itemsH.listEpisodeBrowseMetadata(r.Context(), resultItems)
}()
enrichWG.Wait()
var (
imageURLs map[string]itemListImageURLs
sortMetrics map[string]*sortMetricsResponse
)
store, profileID, _ := h.itemsH.userStoreForRequest(r)
var responseWG sync.WaitGroup
responseWG.Add(2)
go func() {
defer responseWG.Done()
imageURLs = h.itemsH.itemListCardImageURLs(r.Context(), localizedItems)
}()
go func() {
defer responseWG.Done()
sortMetrics = h.itemsH.listSortMetrics(
r.Context(),
resultItems,
sortField,
accessFilter,
overlaySummaries,
store,
apimw.GetUserID(r.Context()),
profileID,
)
}()
responseWG.Wait()
items := make([]itemListResponse, 0, len(localizedItems))
for _, item := range localizedItems {
if item == nil {
continue
}
resp := itemListResponseShell(item, overlaySummaries[item.ContentID], userStates[item.ContentID])
if meta, ok := episodeMetadata[item.ContentID]; ok {
resp.SeriesTitle = meta.SeriesTitle
resp.SeasonNumber = meta.SeasonNumber
resp.EpisodeNumber = meta.EpisodeNumber
}
resp.PosterURL = imageURLs[item.ContentID].posterURL
resp.BackdropURL = imageURLs[item.ContentID].backdropURL
resp.SortMetrics = sortMetrics[item.ContentID]
items = append(items, resp)
}
return items
}
func (h *CatalogHandler) writeCatalogResponse(w http.ResponseWriter, result *catalog.CatalogResult, items []itemListResponse, groupedByWork bool) {
var snapshot string
if !result.SnapshotAt.IsZero() {
snapshot = result.SnapshotAt.Format(time.RFC3339Nano)
}
// A non-empty Provider is the single gate: only the direct-search path sets
// it. Browse / preview / non-relevance-sort q= (which never run a provider)
// and group=work (fresh CatalogResult with empty Provider) all omit it.
var diag *searchDiagnostics
if result.Provider != "" {
diag = &searchDiagnostics{
Provider: result.Provider,
Mode: result.Mode,
SemanticUsed: result.SemanticUsed,
FallbackReason: result.FallbackReason,
IndexPendingUpdates: result.IndexPendingEvents,
}
}
writeJSON(w, http.StatusOK, catalogResponse{
Total: result.Total,
TotalExact: result.TotalExact && !groupedByWork,
HasMore: result.HasMore,
Items: items,
Snapshot: snapshot,
SearchDiagnostics: diag,
})
}
type groupedCatalogEntry struct {
item *models.MediaItem
summary *catalog.WorkSummary
}
func (h *CatalogHandler) resolveGroupedCatalogByWork(r *http.Request, req catalog.CatalogRequest, accessFilter catalog.AccessFilter) (*catalog.CatalogResult, []groupedCatalogEntry, error) {
fetchReq := req
fetchReq.Offset = 0
fetchReq.Limit = groupedCatalogFetchLimit(req.Limit)
fetchReq.SkipTotal = true
seen := map[string]struct{}{}
entries := make([]groupedCatalogEntry, 0, req.Limit+1)
groupIndex := 0
var snapshot time.Time
for {
result, err := h.resolver.Resolve(r.Context(), fetchReq, accessFilter)
if err != nil {
return nil, nil, err
}
if snapshot.IsZero() {
snapshot = result.SnapshotAt
if !snapshot.IsZero() {
fetchReq.SnapshotAt = &snapshot
}
}
summaries, err := h.workSummariesForItems(r, result.Items, accessFilter)
if err != nil {
return nil, nil, err
}
for _, item := range result.Items {
summary := summaries[item.ContentID]
key := groupedCatalogEntryKey(item, summary)
if _, ok := seen[key]; ok {
continue
}
seen[key] = struct{}{}
if groupIndex >= req.Offset {
entries = append(entries, groupedCatalogEntry{item: item, summary: summary})
}
groupIndex++
if len(entries) > req.Limit {
break
}
}
if len(entries) > req.Limit || !result.HasMore || len(result.Items) == 0 {
break
}
fetchReq.Offset += len(result.Items)
}
hasMore := len(entries) > req.Limit
if hasMore {
entries = entries[:req.Limit]
}
total := req.Offset + len(entries)
if hasMore {
total++
}
result := &catalog.CatalogResult{
Items: groupedCatalogItems(entries),
Total: total,
HasMore: hasMore,
TotalExact: false,
SnapshotAt: snapshot,
}
return result, entries, nil
}
func groupedCatalogFetchLimit(limit int) int {
switch {
case limit <= 0:
return 100
case limit >= 100:
return 100
default:
return min(100, limit*4+10)
}
}
func (h *CatalogHandler) workSummariesForItems(r *http.Request, items []*models.MediaItem, filter catalog.AccessFilter) (map[string]*catalog.WorkSummary, error) {
summaries := map[string]*catalog.WorkSummary{}
if h == nil || h.workSummary == nil || len(items) == 0 {
return summaries, nil
}
contentIDs := make([]string, 0, len(items))
seen := map[string]struct{}{}
for _, item := range items {
if !catalogItemCanGroupByWork(item) {
continue
}
if _, ok := seen[item.ContentID]; ok {
continue
}
seen[item.ContentID] = struct{}{}
contentIDs = append(contentIDs, item.ContentID)
}
if len(contentIDs) == 0 {
return summaries, nil
}
if batch, ok := h.workSummary.(catalog.WorkSummaryBatchProvider); ok {
return batch.ListSummariesForContentIDs(r.Context(), contentIDs, filter)
}
for _, contentID := range contentIDs {
summary, err := h.workSummary.GetSummaryForContentID(r.Context(), contentID, filter)
if err != nil {
return nil, err
}
if summary != nil && summary.WorkID != "" {
summaries[contentID] = summary
}
}
return summaries, nil
}
func groupedCatalogItems(entries []groupedCatalogEntry) []*models.MediaItem {
items := make([]*models.MediaItem, 0, len(entries))
for _, entry := range entries {
items = append(items, entry.item)
}
return items
}
func groupedCatalogEntryKey(item *models.MediaItem, summary *catalog.WorkSummary) string {
if catalogItemCanGroupByWork(item) && summary != nil && summary.WorkID != "" {
return "work:" + summary.WorkID
}
if item == nil {
return "item:"
}
return "item:" + item.ContentID
}
func catalogItemCanGroupByWork(item *models.MediaItem) bool {
return item != nil && (item.Type == "ebook" || item.Type == "audiobook")
}
func applyWorkSummaryToCatalogItem(item *itemListResponse, summary *catalog.WorkSummary) {
if item == nil || summary == nil || summary.WorkID == "" {
return
}
item.Type = "work"
item.WorkID = summary.WorkID
item.WorkTitle = summary.Title
item.WorkFormats = summary.Formats
if summary.Title != "" {
item.Title = summary.Title
}
}
func (h *CatalogHandler) HandleGetCatalogFilters(w http.ResponseWriter, r *http.Request) {
if h == nil || h.resolver == nil || h.itemsH == nil {
writeError(w, http.StatusInternalServerError, "internal_error", "Catalog is not configured")
return
}
req, err := catalog.ParseCatalogRequest(r.URL.Query())
if err != nil {
writeError(w, http.StatusBadRequest, "bad_request", err.Error())
return
}
includeTechnical := parseIncludeTechnical(r.URL.Query().Get("include_technical"))
filters, err := h.resolver.ListFiltersWithOptions(
r.Context(),
req,
h.itemsH.accessFilter(r),
catalog.CatalogFilterOptions{IncludeTechnical: includeTechnical},
)
if err != nil {
if errors.Is(err, catalog.ErrInvalidCatalogRequest) {
writeError(w, http.StatusBadRequest, "bad_request", err.Error())
return
}
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to list catalog filters")
return
}
var resolutions *[]string
var audioLanguages *[]string
var subtitleLanguages *[]string
if includeTechnical {
resolutions = &filters.Resolutions
audioLanguages = &filters.AudioLanguages
subtitleLanguages = &filters.SubtitleLanguages
}
writeJSON(w, http.StatusOK, catalogFiltersResponse{
Genres: filters.Genres,
Studios: filters.Studios,
Networks: filters.Networks,
Countries: filters.Countries,
OriginalLanguages: filters.OriginalLanguages,
ContentRatings: filters.ContentRatings,
Authors: filters.Authors,
Narrators: filters.Narrators,
Series: filters.Series,
Resolutions: resolutions,
AudioLanguages: audioLanguages,
SubtitleLanguages: subtitleLanguages,
})
}
// catalogFacetSearchResponse mirrors catalog.CatalogFacetSearchResult on
// the wire. matches[] is always present (empty when no hits); has_more
// is true when the underlying result set held more entries than the
// requested limit.
type catalogFacetSearchResponse struct {
Matches []string `json:"matches"`
HasMore bool `json:"has_more"`
}
// HandleGetCatalogFacetSearch — GET /api/v1/catalog/filters/search
//
// Prefix-typeahead for the high-cardinality filter facets (authors /
// narrators / series, plus genre / studio / network / country /
// original_language / content_rating for consistency). Query
// parameters: same as /api/v1/catalog/filters for scope (source,
// library_id, etc.), plus facet=<name>, q=<prefix>, limit=<N>.
//
// The bulk /api/v1/catalog/filters endpoint stays as the source for
// the initial dropdown render (top 1000 alphabetical); this endpoint
// takes over once the user starts typing.
func (h *CatalogHandler) HandleGetCatalogFacetSearch(w http.ResponseWriter, r *http.Request) {
if h == nil || h.resolver == nil || h.itemsH == nil {
writeError(w, http.StatusInternalServerError, "internal_error", "Catalog is not configured")
return
}
req, err := catalog.ParseCatalogRequest(r.URL.Query())
if err != nil {
writeError(w, http.StatusBadRequest, "bad_request", err.Error())
return
}
facet := strings.TrimSpace(r.URL.Query().Get("facet"))
if facet == "" {
writeError(w, http.StatusBadRequest, "bad_request", "facet parameter is required")
return
}
prefix := r.URL.Query().Get("q")
limit := 20
if raw := strings.TrimSpace(r.URL.Query().Get("limit")); raw != "" {
n, parseErr := strconv.Atoi(raw)
if parseErr != nil || n <= 0 {
writeError(w, http.StatusBadRequest, "bad_request", "limit must be a positive integer")
return
}
if n > 100 {
n = 100
}
limit = n
}
result, err := h.resolver.SearchFacet(
r.Context(),
req,
h.itemsH.accessFilter(r),
facet,
prefix,
limit,
)
if err != nil {
if errors.Is(err, catalog.ErrInvalidCatalogRequest) {
writeError(w, http.StatusBadRequest, "bad_request", err.Error())
return
}
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to search catalog facet")
return
}
matches := result.Matches
if matches == nil {
matches = []string{}
}
writeJSON(w, http.StatusOK, catalogFacetSearchResponse{
Matches: matches,
HasMore: result.HasMore,
})
}
func parseIncludeTechnical(raw string) bool {
if strings.TrimSpace(raw) == "" {
return true
}
includeTechnical, err := strconv.ParseBool(raw)
if err != nil {
return true
}
return includeTechnical
}
func (h *CatalogHandler) HandlePostCatalogQuery(w http.ResponseWriter, r *http.Request) {
if h == nil || h.itemsH == nil || h.itemsH.browseRepo == nil {
writeError(w, http.StatusInternalServerError, "internal_error", "Catalog is not configured")
return
}
var req filterItemsRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "bad_request", "Invalid request body")
return
}
if req.Limit <= 0 {
req.Limit = 20
}
if req.Limit > 100 {
req.Limit = 100
}
fb := sections.NewFilterBuilder("mi")
filterWhere, filterArgs, err := fb.Build(req.FilterConfig)
if err != nil {
writeError(w, http.StatusBadRequest, "bad_request", "Invalid filter: "+err.Error())
return
}
accessFilter := h.itemsH.accessFilter(r)
if req.LibraryID > 0 {
accessFilter.PresentationLibraryID = &req.LibraryID
}
libraryIDs := accessFilter.AllowedLibraryIDs
var conditions []string
args := filterArgs
argIdx := fb.ArgIdx()
if filterWhere != "" {
conditions = append(conditions, filterWhere)
}
disabledLibraryIDs := accessFilter.DisabledLibraryIDs
fromClause := "media_items mi"
if libraryIDs != nil || req.LibraryID > 0 || len(disabledLibraryIDs) > 0 {
fromClause = "media_items mi JOIN media_item_libraries mil ON mi.content_id = mil.content_id"
}
if req.LibraryID > 0 {
conditions = append(conditions, fmt.Sprintf("mil.media_folder_id = $%d", argIdx))
args = append(args, req.LibraryID)
argIdx++
}
if libraryIDs != nil {
if len(libraryIDs) == 0 {
writeJSON(w, http.StatusOK, browseResponse{Items: []itemListResponse{}, Total: 0})
return
}
conditions = append(conditions, fmt.Sprintf("mil.media_folder_id = ANY($%d)", argIdx))
args = append(args, libraryIDs)
argIdx++
}
if len(disabledLibraryIDs) > 0 {
conditions = append(conditions, fmt.Sprintf("NOT (mil.media_folder_id = ANY($%d))", argIdx))
args = append(args, disabledLibraryIDs)
argIdx++
}
catalog.ApplySectionAccessFilter("mi", catalog.AccessFilter{MaxContentRating: accessFilter.MaxContentRating}, &conditions, &args, &argIdx)
whereClause := ""
if len(conditions) > 0 {
whereClause = "WHERE " + strings.Join(conditions, " AND ")
}
sortClause := "ORDER BY mi.created_at DESC"
if req.Sort != "" {
sortClause = filterSortClause(req.Sort, req.Order)
}
// Single-pass query: COUNT(*) OVER () returns the same total on every
// row of the filtered set, so we read it once from the first scanned row
// instead of firing a separate SELECT COUNT(*).
query := fmt.Sprintf(
`SELECT %s, COUNT(*) OVER () AS total_count FROM %s %s %s LIMIT $%d OFFSET $%d`,
filterItemColumns("mi"), fromClause, whereClause, sortClause, argIdx, argIdx+1,
)
args = append(args, req.Limit, req.Offset)
rows, err := h.itemsH.browseRepo.Pool().Query(r.Context(), query, args...)
if err != nil {
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to query items")
return
}
defer rows.Close()
var total int
modelItems := make([]*models.MediaItem, 0)
for rows.Next() {
var item models.MediaItem
var rowTotal int
scanErr := rows.Scan(
&item.ContentID, &item.Type, &item.Title, &item.SortTitle, &item.OriginalTitle,
&item.Year, &item.Genres, &item.ContentRating, &item.Runtime, &item.Overview, &item.Tagline,
&item.RatingIMDB, &item.RatingTMDB, &item.RatingRTCritic, &item.RatingRTAudience,
&item.ImdbID, &item.TmdbID, &item.TvdbID,
&item.PosterPath, &item.PosterThumbhash, &item.BackdropPath, &item.BackdropThumbhash, &item.LogoPath,
&item.MetadataS3Path, &item.MetadataEtag, &item.SeasonCount,
&item.Studios, &item.Networks, &item.Countries, &item.FirstAirDate, &item.LastAirDate,
&item.MatchedAt, &item.Status, &item.CreatedAt, &item.UpdatedAt,
&rowTotal,
)
if scanErr != nil {
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to scan item")
return
}
modelItems = append(modelItems, &item)
total = rowTotal
}
// COUNT(*) OVER () emits no rows when the data SELECT is empty, so total
// stays 0 even when the broader result set has matching rows (e.g. OFFSET
// past the last page). Re-query the count to give callers the real total.
// Skip when offset == 0 because in that case an empty page genuinely means
// total = 0. Mirrors the fallback in browse.go / query_executor.go /
// favorites_browse.go / item_repo.go Search.
if len(modelItems) == 0 && req.Offset > 0 {
countQuery := fmt.Sprintf(
"SELECT COUNT(*) FROM (SELECT 1 FROM %s %s) sub",
fromClause, whereClause,
)
// Drop the trailing limit, offset args from the data query.
countArgs := args[:len(args)-2]
if err := h.itemsH.browseRepo.Pool().QueryRow(r.Context(), countQuery, countArgs...).Scan(&total); err != nil {
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to count items")
return
}
}
userStates := h.itemsH.listItemUserStates(r, modelItems)
items := make([]itemListResponse, 0, len(modelItems))
for _, item := range modelItems {
if h.itemsH.detailSvc != nil {
if localized, locErr := h.itemsH.detailSvc.LocalizeItemModel(r.Context(), item, accessFilter); locErr == nil && localized != nil {
item = localized
}
}
items = append(items, h.itemsH.toItemListResponseWithOverlay(r, item, nil, userStates[item.ContentID]))
}
writeJSON(w, http.StatusOK, browseResponse{
Total: total,
HasMore: req.Offset+req.Limit < total,
Items: items,
})
}
func (h *CatalogHandler) HandleLegacySearch(w http.ResponseWriter, r *http.Request) {
if h == nil || h.itemsH == nil || h.itemsH.itemRepo == nil {
writeError(w, http.StatusInternalServerError, "internal_error", "Catalog is not configured")
return
}
if values, ok := buildLegacySearchCatalogValues(r.URL.Query()); ok && h.resolver != nil {
h.itemsH.writeCatalogBrowseResponse(w, r, values)
return
}
query := strings.TrimSpace(r.URL.Query().Get("q"))
if query == "" {
writeError(w, http.StatusBadRequest, "bad_request", "Search query 'q' is required")
return
}
limit := catalog.ParseIntParam(r.URL.Query().Get("limit"))
if limit <= 0 {
limit = 20
}
if limit > 100 {
limit = 100
}
offset := max(catalog.ParseIntParam(r.URL.Query().Get("offset")), 0)
items, total, err := h.itemsH.itemRepo.Search(r.Context(), query, parseSearchTypes(r.URL.Query()["type"]), limit, offset, h.itemsH.accessFilter(r))
if err != nil {
slog.ErrorContext(r.Context(), "search failed", "component", "api", "query", query, "error", err)
writeError(w, http.StatusInternalServerError, "internal_error", "Search failed")
return
}
userStates := h.itemsH.listItemUserStates(r, items)
resp := make([]itemListResponse, 0, len(items))
for _, item := range items {
resp = append(resp, h.itemsH.toItemListResponseWithOverlay(r, item, nil, userStates[item.ContentID]))
}
writeJSON(w, http.StatusOK, browseResponse{
Total: total,
HasMore: offset+len(resp) < total,
Items: resp,
})
}