thread-master/apps/backend/internal/module/scout/usecase/search_pipeline.go

package usecase

import (
	"context"
	"fmt"
	"strings"

	"apps/backend/internal/module/scout/domain"
)

const maxPipelineRawCandidates = 320

type SearchPipelineDiagnostics struct {
	RawCount          int
	DuplicateCount    int
	IrrelevantCount   int
	EligibleCount     int
	Stages            int
	SourceUnavailable bool
	ShortfallReasons  []string
}

type SearchPipelineResult struct {
	Hits        []ThreadSearchResult
	Diagnostics SearchPipelineDiagnostics
}

type searchPipelineRunner struct {
	ctx          context.Context
	service      *Service
	brief        *domain.RunBrief
	terms        []string
	target       int
	path         string
	storageState string
	source       func(context.Context, string, int) ([]ThreadSearchResult, error)
	secondary    func(context.Context, string, int) ([]ThreadSearchResult, error)
	evaluator    *CandidateEvaluator
	hits         []ThreadSearchResult
	diagnostics  SearchPipelineDiagnostics
	lastErr      error
	sourceOK     bool
}

// searchEligiblePipeline executes initial, same-source boost and optional
// secondary-source stages. It evaluates each raw candidate immediately, so a
// raw hit count can never satisfy target before relevance and dedupe checks.
func (s *Service) searchEligiblePipeline(
	ctx context.Context,
	ownerUID int64,
	terms []string,
	brief *domain.RunBrief,
	target int,
	path string,
	storageState string,
	initialPerQuery int,
) (SearchPipelineResult, error) {
	if s == nil || brief == nil || len(terms) == 0 {
		return SearchPipelineResult{}, fmt.Errorf("%w: search pipeline input required", domain.ErrValidation)
	}
	if initialPerQuery < 1 {
		initialPerQuery = 1
	}
	if initialPerQuery > 20 {
		initialPerQuery = 20
	}
	runner := &searchPipelineRunner{
		ctx: ctx, service: s, brief: brief, terms: nonEmptyTerms(terms), target: target,
		path: path, storageState: storageState, evaluator: NewCandidateEvaluator(brief),
	}
	runner.evaluator.SetHistoricalSeenLookup(func(identity string) (bool, error) {
		return s.Repo.HasSeenIdentity(ctx, ownerUID, identity)
	})
	if path == domain.PathCrawler && s.Crawler != nil && storageState != "" {
		runner.source = func(ctx context.Context, term string, limit int) ([]ThreadSearchResult, error) {
			return s.Crawler.SearchChrome(ctx, storageState, []string{term}, limit)
		}
		runner.secondary = func(ctx context.Context, term string, limit int) ([]ThreadSearchResult, error) {
			if s.Provider == nil {
				return nil, fmt.Errorf("search provider is not configured")
			}
			return s.Provider.SearchThreads(ctx, []string{term}, limit)
		}
	} else if s.Provider != nil {
		runner.source = func(ctx context.Context, term string, limit int) ([]ThreadSearchResult, error) {
			return s.Provider.SearchThreads(ctx, []string{term}, limit)
		}
	}
	if runner.source == nil {
		return SearchPipelineResult{}, fmt.Errorf("%w: primary search source unavailable", domain.ErrValidation)
	}

	runner.runStage(initialPerQuery, runner.source)
	if target <= 0 || runner.reachedTarget() || runner.diagnostics.RawCount >= maxPipelineRawCandidates {
		return runner.finish()
	}
	runner.runStage(20, runner.source)
	if runner.reachedTarget() || runner.diagnostics.RawCount >= maxPipelineRawCandidates {
		return runner.finish()
	}
	// API providers such as Exa do not expose an offset in this adapter, so a
	// second identical request can return the same first page. When the user
	// approved only one term, use a small, intent-derived set of conjunctions
	// before giving up or switching source. The evaluator still applies the
	// original topic signature, so expansion increases recall without relaxing
	// relevance.
	if expansionTerms := searchPipelineExpansionTerms(runner.brief, runner.terms); len(expansionTerms) > 0 {
		runner.runStageForTerms(20, runner.source, expansionTerms)
		if runner.reachedTarget() || runner.diagnostics.RawCount >= maxPipelineRawCandidates {
			return runner.finish()
		}
	}
	if runner.secondary != nil {
		runner.runStage(20, runner.secondary)
	}
	return runner.finish()
}

func (r *searchPipelineRunner) runStage(perQuery int, source func(context.Context, string, int) ([]ThreadSearchResult, error)) {
	r.runStageForTerms(perQuery, source, r.terms)
}

func (r *searchPipelineRunner) runStageForTerms(perQuery int, source func(context.Context, string, int) ([]ThreadSearchResult, error), terms []string) {
	if source == nil || r.reachedTarget() || r.diagnostics.RawCount >= maxPipelineRawCandidates {
		return
	}
	r.diagnostics.Stages++
	for _, term := range terms {
		if r.reachedTarget() || r.diagnostics.RawCount >= maxPipelineRawCandidates {
			break
		}
		remaining := maxPipelineRawCandidates - r.diagnostics.RawCount
		limit := perQuery
		if limit > remaining {
			limit = remaining
		}
		hits, err := source(r.ctx, term, limit)
		if err != nil {
			r.lastErr = err
			r.diagnostics.SourceUnavailable = true
		}
		if err == nil {
			r.sourceOK = true
		}
		if len(hits) > remaining {
			hits = hits[:remaining]
		}
		r.diagnostics.RawCount += len(hits)
		for _, hit := range hits {
			if hit.MatchedQuery == "" {
				hit.MatchedQuery = term
			}
			if permalink := canonicalPermalink(hit.URL); permalink != "" {
				hit.URL = permalink
			}
			evaluation := r.evaluator.EvaluateAt(hit, domain.NowNano())
			switch evaluation.Decision {
			case CandidateEligible:
				r.hits = append(r.hits, hit)
			case CandidateDuplicate:
				r.diagnostics.DuplicateCount++
			default:
				r.diagnostics.IrrelevantCount++
			}
			if r.reachedTarget() {
				break
			}
		}
	}
}

func searchPipelineExpansionTerms(brief *domain.RunBrief, selected []string) []string {
	if brief == nil || brief.Mode != domain.ModeActivity || len(nonEmptyTerms(selected)) != 1 {
		return nil
	}
	// Keep this bounded: it is a recovery stage for a one-term approval, not a
	// second unrestricted planner. planActivityTerms is deterministic and uses
	// the same Threads short-query contract as the review UI.
	planned := planActivityTerms(brief.Intent)
	selectedKey := normalizeTopicText(selected[0])
	out := make([]string, 0, 4)
	seen := map[string]struct{}{selectedKey: {}}
	for _, term := range planned {
		term = strings.TrimSpace(term)
		key := normalizeTopicText(term)
		if key == "" {
			continue
		}
		if _, exists := seen[key]; exists {
			continue
		}
		if len(semanticTopicParts(term)) == 0 {
			continue
		}
		seen[key] = struct{}{}
		out = append(out, term)
		if len(out) == 4 {
			break
		}
	}
	return out
}

func (r *searchPipelineRunner) reachedTarget() bool {
	return r.target > 0 && len(r.hits) >= r.target
}

func (r *searchPipelineRunner) finish() (SearchPipelineResult, error) {
	r.diagnostics.EligibleCount = len(r.hits)
	if r.target > 0 && len(r.hits) < r.target {
		r.diagnostics.ShortfallReasons = shortfallReasons(r.diagnostics)
	}
	if !r.sourceOK && r.lastErr != nil {
		return SearchPipelineResult{Hits: r.hits, Diagnostics: r.diagnostics}, r.lastErr
	}
	return SearchPipelineResult{Hits: r.hits, Diagnostics: r.diagnostics}, nil
}

func shortfallReasons(d SearchPipelineDiagnostics) []string {
	var reasons []string
	if d.SourceUnavailable {
		reasons = append(reasons, domain.ShortfallSourceUnavailable)
	}
	if d.DuplicateCount > 0 {
		reasons = append(reasons, domain.ShortfallDuplicateExhausted)
	}
	if d.IrrelevantCount > 0 {
		reasons = append(reasons, domain.ShortfallRelevanceExhausted)
	}
	if d.RawCount >= maxPipelineRawCandidates {
		reasons = append(reasons, domain.ShortfallLimitReached)
	}
	if len(reasons) == 0 {
		reasons = []string{domain.ShortfallSourceExhausted}
	}
	return reasons
}

func normalizePipelineTerms(terms []string) []string {
	out := make([]string, 0, len(terms))
	for _, term := range terms {
		if term = strings.TrimSpace(term); term != "" {
			out = append(out, term)
		}
	}
	return dedupeTerms(out)
}