package usecase import ( "context" "sort" "apps/backend/internal/module/radar/domain" ) type scoredCandidate struct { cand *domain.CandidatePost result *JudgeResult credits int } /* ProcessCandidates judges hits, applies daily quota truncation by score, and persists. - Already judged external_ids (resume) are skipped without re-billing. - Existing external_id → merge matched_terms only. - Incomplete reasons → skip persist, count as judge failure on sweep. - Truncation is success path: truncated_count increments, not an error. */ func (s *Service) ProcessCandidates( ctx context.Context, ownerUID int64, watch *domain.RadarWatch, profile *domain.ServiceProfile, sweepID string, cands []*domain.CandidatePost, alreadyJudged map[string]bool, ) (created, rematched, judged, truncated, failed int, credits int, err error) { matchEvaluated, matchMerged, fitRejected, budgetDeferred := 0, 0, 0, 0 if alreadyJudged == nil { alreadyJudged = map[string]bool{} } maxDaily, err := s.MaxDailyOpportunities(ctx, ownerUID) if err != nil { return 0, 0, 0, 0, 0, 0, err } todayCount, err := s.Repo.CountToday(ctx, ownerUID, domain.NowNano()) if err != nil { return 0, 0, 0, 0, 0, 0, err } remaining := maxDaily - int(todayCount) if remaining < 0 { remaining = 0 } // Cap how many we judge this run: remaining + small buffer so we can rank then truncate. // Spec: judge count hard-capped by daily max. judgeBudget := remaining if judgeBudget <= 0 { // Existing opportunities may still receive a previously unseen product // match; only genuinely new opportunities are quota-truncated. for _, c := range cands { if c == nil || c.ExternalID == "" || alreadyJudged[c.ExternalID] { continue } existing, gerr := s.Repo.GetByExternalID(ctx, ownerUID, c.ExternalID) if gerr == nil && existing != nil && watch != nil && watch.ContextMode == domain.WatchContextProduct && productMatchFor(existing, watch.ProductID) == nil { _, _ = s.Repo.UpsertByExternalID(ctx, &domain.Opportunity{OwnerUID: ownerUID, ExternalID: c.ExternalID, MatchedTerms: []string{c.MatchedTerm}}) matchEvaluated++ pcredits, perr := s.mergeExistingProductCandidate(ctx, ownerUID, watch, c, existing) credits += pcredits judged++ rematched++ if perr == nil { matchMerged++ } else { failed++ } continue } truncated++ budgetDeferred++ } if sweepID != "" { _, _ = s.Repo.UpdateSweep(ctx, sweepID, domain.SweepDelta{JudgedCount: judged, TruncatedCount: truncated, BudgetDeferredCount: budgetDeferred, CreditsUsed: credits, CreditJudge: credits, MatchEvaluatedCount: matchEvaluated, MatchMergedCount: matchMerged, FitRejectedCount: fitRejected}) } return 0, rematched, judged, truncated, failed, credits, nil } var scored []scoredCandidate var judgedIDs []string checkpoint := func() { if sweepID != "" && len(judgedIDs)%10 == 0 && len(judgedIDs) > 0 { _, _ = s.Repo.UpdateSweep(ctx, sweepID, domain.SweepDelta{JudgedExternalIDs: judgedIDs}) } } for _, c := range cands { if c == nil || c.ExternalID == "" { continue } if alreadyJudged[c.ExternalID] { continue } // Dedupe check without re-judge: if exists, merge term only. existing, gerr := s.Repo.GetByExternalID(ctx, ownerUID, c.ExternalID) if gerr == nil && existing != nil { term := c.MatchedTerm _, _ = s.Repo.UpsertByExternalID(ctx, &domain.Opportunity{ OwnerUID: ownerUID, ExternalID: c.ExternalID, MatchedTerms: []string{term}, }) if watch != nil && watch.ContextMode == domain.WatchContextProduct && productMatchFor(existing, watch.ProductID) == nil { matchEvaluated++ } if pcredits, perr := s.mergeExistingProductCandidate(ctx, ownerUID, watch, c, existing); perr != nil { credits += pcredits failed++ } else { credits += pcredits rematched++ if watch != nil && watch.ContextMode == domain.WatchContextProduct && productMatchFor(existing, watch.ProductID) == nil { matchMerged++ } } judgedIDs = append(judgedIDs, c.ExternalID) judged++ checkpoint() continue } if len(scored) >= judgeBudget { truncated++ budgetDeferred++ continue } if watch != nil && watch.ContextMode == domain.WatchContextProduct { matchEvaluated++ } res, cred, jerr := s.JudgeCandidate(ctx, ownerUID, profile, watch, c) credits += cred if jerr != nil || res == nil { failed++ judgedIDs = append(judgedIDs, c.ExternalID) judged++ checkpoint() continue } if res.ProductMatch != nil && (!res.ProductMatch.Eligible || res.ProductMatch.Excluded) { fitRejected++ } scored = append(scored, scoredCandidate{cand: c, result: res, credits: cred}) judgedIDs = append(judgedIDs, c.ExternalID) judged++ checkpoint() } // Rank qualified/rejected by score desc; always persist rejected; qualified subject to remaining. sort.SliceStable(scored, func(i, j int) bool { return scored[i].result.IntentScore > scored[j].result.IntentScore }) for _, sc := range scored { res := sc.result // rejected always stored if reasons ok if res.Status == domain.OppRejected { inserted, perr := s.persistOne(ctx, ownerUID, watch, sc.cand, res) if perr != nil { failed++ continue } if inserted { created++ } else { rematched++ } continue } if remaining <= 0 { truncated++ budgetDeferred++ continue } inserted, perr := s.persistOne(ctx, ownerUID, watch, sc.cand, res) if perr != nil { failed++ continue } if inserted { created++ } else { rematched++ } remaining-- } if sweepID != "" { delta := domain.SweepDelta{ JudgedCount: judged, CreatedCount: created, TruncatedCount: truncated, CreditsUsed: credits, CreditJudge: credits, JudgedExternalIDs: judgedIDs, MatchEvaluatedCount: matchEvaluated, MatchMergedCount: matchMerged, FitRejectedCount: fitRejected, BudgetDeferredCount: budgetDeferred, } if _, uerr := s.Repo.UpdateSweep(ctx, sweepID, delta); uerr != nil { return created, rematched, judged, truncated, failed, credits, uerr } } return created, rematched, judged, truncated, failed, credits, nil } func (s *Service) persistOne(ctx context.Context, ownerUID int64, watch *domain.RadarWatch, cand *domain.CandidatePost, res *JudgeResult) (inserted bool, err error) { if err := domain.ValidateReasons(res.Reasons); err != nil { return false, err } watchID := "" if watch != nil { watchID = watch.ID } priority := scoreOpportunityPriority(cand, res) o := &domain.Opportunity{ ID: domain.NewID(), OwnerUID: ownerUID, WatchID: watchID, Source: domain.OppSourceThreads, ExternalID: cand.ExternalID, Permalink: cand.Permalink, AuthorHandle: cand.AuthorHandle, Text: cand.Text, PostedAt: cand.PostedAt, Status: res.Status, IntentScore: res.IntentScore, IntentBand: res.IntentBand, Reasons: res.Reasons, RegionDetected: res.RegionDetected, RegionMatch: res.RegionMatch, FreshnessHours: res.FreshnessHours, MatchedService: res.MatchedService, MatchedTerms: []string{cand.MatchedTerm}, RejectReason: res.RejectReason, ProductMatches: nil, PriorityScore: priority.Priority, PriorityBand: priority.Band, PainFitScore: priority.PainFit, DemandIntentScore: priority.DemandIntent, EvidenceQualityScore: priority.EvidenceQuality, FreshnessScore: priority.Freshness, DemandEvidence: priority.Evidence, LastMatchedAt: domain.NowNano(), } if watch != nil && watch.ContextMode == domain.WatchContextProduct { if dm, derr := s.GetDemandMap(ctx, ownerUID, watch.ProductID); derr == nil && dm != nil { o.DemandInputVersion, o.DemandMapVersion = dm.DemandInputVersion, dm.MapVersion } } if res.ProductMatch != nil { o.ProductMatches = []*domain.ProductMatch{domain.CloneProductMatch(res.ProductMatch)} } if err := mergeProductMatchIntoOpportunity(o, res.ProductMatch); res.ProductMatch != nil && err != nil { return false, err } if o.IntentBand == "" { o.ApplyBandFromScore() } existing, gerr := s.Repo.GetByExternalID(ctx, ownerUID, cand.ExternalID) already := gerr == nil && existing != nil if _, err := s.Repo.UpsertByExternalID(ctx, o); err != nil { return false, err } return !already, nil }