thread-master/apps/backend/internal/module/radar/usecase/sweep_run.go

package usecase

import (
	"context"
	"fmt"
	"strings"

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

// SweepRunResult is the outcome of one radar_sweep job execution.
type SweepRunResult struct {
	Sweep        *domain.RadarSweep
	Created      int
	Rematched    int
	Judged       int
	Truncated    int
	FailedJudges int
	FetchFailed  bool
	FailedReason string
}

// Notifier sends in-app alerts for sweep failures.
type SweepNotifier interface {
	NotifySweepFailed(ctx context.Context, ownerUID int64, sweepID, watchID, reason string) error
}

/*
RunSweep is the full pipeline: create/reuse Sweep → fetch → judge → persist → finish.

Fetch failure: Sweep failed_reason set, job should fail, notify user — never empty success.
Partial judge failures: Sweep still succeeds with counters.
Resume: skip external_ids already on the Sweep record.
*/
func (s *Service) RunSweep(ctx context.Context, ownerUID int64, watchID, jobID string) (*SweepRunResult, error) {
	if ownerUID <= 0 || watchID == "" {
		return nil, fmt.Errorf("%w: owner_uid and watch_id required", domain.ErrValidation)
	}
	w, err := s.Repo.GetWatch(ctx, watchID)
	if err != nil {
		return nil, err
	}
	if w.OwnerUID != ownerUID {
		return nil, domain.ErrForbidden
	}
	if w.Status != domain.WatchActive {
		return nil, fmt.Errorf("%w: only active watches can be swept (status=%s)", domain.ErrValidation, w.Status)
	}

	var profile *domain.ServiceProfile
	var productContext *ProductContextSnapshot
	if w.ContextMode == domain.WatchContextProduct {
		productContext, err = s.LoadProductContext(ctx, ownerUID, w.BrandID, w.ProductID)
		if err != nil {
			reason := "產品上下文不可用，已暫停巡邏：" + err.Error()
			_, _ = s.PauseProductUnavailable(ctx, ownerUID, watchID, domain.PauseReasonProductUnavailable)
			_ = s.notifySweepFailed(ctx, ownerUID, "", watchID, reason)
			return nil, err
		}
		// Product watches do not require a service profile; it remains optional
		// for regional/freshness context when present.
		profile, _ = s.Repo.GetServiceProfile(ctx, ownerUID)
	} else {
		profile, err = s.Repo.GetServiceProfile(ctx, ownerUID)
		if err != nil {
			return nil, fmt.Errorf("%w: service profile required for sweep", domain.ErrValidation)
		}
	}

	// Resume: if a sweep already exists for this job, reuse it.
	var sw *domain.RadarSweep
	if jobID != "" {
		if existing, gerr := s.Repo.GetSweepByJobID(ctx, jobID); gerr == nil && existing != nil {
			sw = existing
		}
	}
	if sw == nil {
		sw, err = s.BeginSweepRecord(ctx, ownerUID, watchID, jobID, domain.SweepPathAPI)
		if err != nil {
			return nil, err
		}
	}

	already := map[string]bool{}
	for _, id := range sw.JudgedExternalIDs {
		already[id] = true
	}

	fetchWatch := mergeFetchWatch(w, nil)
	if w.ContextMode == domain.WatchContextProduct {
		if plan, perr := s.BuildProductQueryPlan(ctx, ownerUID, w.ProductID); perr == nil && plan != nil && len(plan.Groups) > 0 {
			fetchWatch = mergeFetchWatch(w, plan)
		}
	}
	cands, path, fetchCredits, ferr := s.FetchCandidates(ctx, ownerUID, fetchWatch, 40)
	if ferr != nil {
		reason := humanFetchError(ferr)
		end := domain.NowNano()
		_, _ = s.Repo.UpdateSweep(ctx, sw.ID, domain.SweepDelta{
			FailedReason: &reason,
			EndedAt:      end,
			CreditsUsed:  fetchCredits,
		})
		_ = s.notifySweepFailed(ctx, ownerUID, sw.ID, watchID, reason)
		sw, _ = s.Repo.GetSweep(ctx, sw.ID)
		return &SweepRunResult{Sweep: sw, FetchFailed: true, FailedReason: reason}, ferr
	}

	rawHitCount := len(cands)
	filtered, prefilter := PrefilterCandidates(cands, w, productContext)
	cands = filtered
	_, _ = s.Repo.UpdateSweep(ctx, sw.ID, domain.SweepDelta{
		HitCount:               rawHitCount,
		CreditsUsed:            fetchCredits,
		CreditSearch:           fetchCredits,
		DedupedCount:           prefilter.Deduped,
		PrefilterPassCount:     prefilter.Pass,
		PrefilterReviewCount:   prefilter.Review,
		PrefilterRejectedCount: prefilter.Rejected,
	})
	// set path on record
	if path != "" && sw.Path != path {
		sw.Path = path
		// path is not in SweepDelta; re-save via create is wrong — store via failed empty update isn't enough.
		// Use UpdateSweep only for counters; path was set at Begin — recreate if needed.
		if sw.Path == domain.SweepPathAPI && path == domain.SweepPathCrawler {
			// best-effort: include in failed_reason empty path note not needed; set via full get/update memory
			_ = s.setSweepPath(ctx, sw.ID, path)
		}
	}

	created, rematched, judged, truncated, failed, judgeCredits, perr := s.ProcessCandidates(
		ctx, ownerUID, w, profile, sw.ID, cands, already,
	)
	if perr != nil {
		reason := "判定流程失敗：" + perr.Error()
		end := domain.NowNano()
		_, _ = s.Repo.UpdateSweep(ctx, sw.ID, domain.SweepDelta{FailedReason: &reason, EndedAt: end})
		_ = s.notifySweepFailed(ctx, ownerUID, sw.ID, watchID, reason)
		return nil, perr
	}

	end := domain.NowNano()
	var failPtr *string
	if failed > 0 && created == 0 && judged > 0 {
		r := fmt.Sprintf("%d 筆判定失敗", failed)
		failPtr = &r
	}
	sw, _ = s.Repo.UpdateSweep(ctx, sw.ID, domain.SweepDelta{
		CreditsUsed:  judgeCredits,
		EndedAt:      end,
		FailedReason: failPtr,
	})
	_ = s.Repo.TouchWatchSweptAt(ctx, watchID, end)

	return &SweepRunResult{
		Sweep:        sw,
		Created:      created,
		Rematched:    rematched,
		Judged:       judged,
		Truncated:    truncated,
		FailedJudges: failed,
	}, nil
}

func (s *Service) setSweepPath(ctx context.Context, id, path string) error {
	sw, err := s.Repo.GetSweep(ctx, id)
	if err != nil {
		return err
	}
	sw.Path = path
	// Memory/mongo lack ReplaceSweep — use UpdateSweep no-op + store path only on create.
	// For mongo, UpdateOne $set path:
	type pathSetter interface {
		SetSweepPath(ctx context.Context, id, path string) error
	}
	if ps, ok := s.Repo.(pathSetter); ok {
		return ps.SetSweepPath(ctx, id, path)
	}
	// memory: mutate via UpdateSweep zero + re-get won't change path; patch memory map if possible
	if m, ok := s.Repo.(interface {
		PatchSweepPath(ctx context.Context, id, path string) error
	}); ok {
		return m.PatchSweepPath(ctx, id, path)
	}
	_ = sw
	return nil
}

func (s *Service) notifySweepFailed(ctx context.Context, ownerUID int64, sweepID, watchID, reason string) error {
	if s.Notifier == nil {
		return nil
	}
	return s.Notifier.NotifySweepFailed(ctx, ownerUID, sweepID, watchID, reason)
}

// mergeFetchWatch keeps the subscriber's own keywords and adds demand-map
// queries. Replacing the watch terms with only plan groups made patrols
// miss the phrases the user actually typed.
func mergeFetchWatch(w *domain.RadarWatch, plan *domain.QueryPlan) *domain.RadarWatch {
	if w == nil {
		return nil
	}
	planned := *w
	terms := append([]string{}, w.Terms...)
	excludes := append([]string{}, w.ExcludeTerms...)
	seen := map[string]bool{}
	for _, t := range terms {
		seen[strings.ToLower(strings.TrimSpace(t))] = true
	}
	if plan != nil {
		for _, group := range plan.Groups {
			q := strings.TrimSpace(group.Query)
			if q != "" && !seen[strings.ToLower(q)] {
				terms = append(terms, q)
				seen[strings.ToLower(q)] = true
			}
			excludes = append(excludes, group.Exclude...)
		}
	}
	planned.Terms = domain.ExpandSearchTerms(terms, domain.MaxWatchTerms)
	planned.ExcludeTerms = excludes
	return &planned
}

func humanFetchError(err error) string {
	if err == nil {
		return "抓取失敗"
	}
	msg := err.Error()
	low := strings.ToLower(msg)
	// never include token-like blobs
	if strings.Contains(low, "bearer ") {
		return "抓取路徑不可用"
	}
	if strings.Contains(low, "crawler session expired") ||
		strings.Contains(low, "crawler session is invalid") ||
		strings.Contains(low, "crawler session required") {
		return "今天沒巡到：Chrome 登入已過期。請到設定重新同步已登入的 Threads 分頁，或先關掉開發模式改走 API 搜尋。"
	}
	if len(msg) > 200 {
		msg = msg[:200]
	}
	return "今天沒巡到：" + msg
}