thread-master/backend/internal/worker/job/scan_viral.go

package job

import (
	"context"
	"fmt"
	"strings"

	"haixun-backend/internal/library/clock"
	app "haixun-backend/internal/library/errors"
	"haixun-backend/internal/library/errors/code"
	"haixun-backend/internal/library/placement"
	libviral "haixun-backend/internal/library/viral"
	"haixun-backend/internal/library/websearch"
	domai "haixun-backend/internal/model/ai/domain/usecase"
	aiusecase "haixun-backend/internal/model/ai/usecase"
	copydraftusecase "haixun-backend/internal/model/copy_draft/domain/usecase"
	missionentity "haixun-backend/internal/model/copy_mission/domain/entity"
	missiondomain "haixun-backend/internal/model/copy_mission/domain/usecase"
	jobdom "haixun-backend/internal/model/job/domain/usecase"
	personaentity "haixun-backend/internal/model/persona/domain/entity"
	personadomain "haixun-backend/internal/model/persona/domain/usecase"
	placementusecase "haixun-backend/internal/model/placement/usecase"
	scanpostusecase "haixun-backend/internal/model/scan_post/domain/usecase"
	accountentity "haixun-backend/internal/model/threads_account/domain/entity"
	threadsaccountdomain "haixun-backend/internal/model/threads_account/domain/usecase"
)

type ScanViralDeps struct {
	Jobs           jobdom.UseCase
	CopyMission    missiondomain.UseCase
	Persona        personadomain.UseCase
	ScanPost       scanpostusecase.UseCase
	CopyDraft      copydraftusecase.UseCase
	ThreadsAccount threadsaccountdomain.UseCase
	Placement      placementusecase.UseCase
	AI             aiusecase.UseCase
}

func RegisterScanViralHandler(runner *Runner, deps ScanViralDeps) {
	if runner == nil {
		return
	}
	runner.RegisterStepHandler("viral_crawl", func(ctx context.Context, step StepContext) error {
		return runScanViral(ctx, step, deps)
	})
}

func runScanViral(ctx context.Context, step StepContext, deps ScanViralDeps) error {
	payload := step.Run.Payload
	tenantID, ownerUID := runActorFromPayload(payload, step.Run)
	personaID := personaIDFromPayload(payload)
	missionID := copyMissionIDFromPayload(payload)
	if tenantID == "" || ownerUID == "" || personaID == "" {
		return fmt.Errorf("scan-viral payload missing tenant_id, owner_uid, or persona_id")
	}

	persona, err := deps.Persona.Get(ctx, tenantID, ownerUID, personaID)
	if err != nil {
		return err
	}
	var mission *missiondomain.MissionSummary
	if missionID != "" {
		item, err := deps.CopyMission.Get(ctx, tenantID, ownerUID, personaID, missionID)
		if err != nil {
			return err
		}
		mission = item
	}

	research, err := deps.Placement.ResearchSettings(ctx, tenantID, ownerUID)
	if err != nil {
		return err
	}
	memberCtx, err := deps.ThreadsAccount.ResolveMemberPlacementContext(ctx, tenantID, ownerUID, research)
	if err != nil {
		return err
	}
	if !memberCtx.HasDiscoverPath() {
		return fmt.Errorf("爆款掃描需要 Threads API、Chrome Session 或 Web Search API（請檢查連線模式與搜尋來源）")
	}

	updateProgress := func(summary string, percentage int) {
		_ = step.Heartbeat(ctx)
		_, _ = deps.Jobs.UpdateProgress(ctx, jobdom.UpdateProgressRequest{
			JobID:      step.JobID,
			WorkerID:   step.WorkerID,
			Phase:      "viral_crawl",
			Summary:    summary,
			Percentage: percentage,
		})
	}

	bootstrap := boolField(payload, "bootstrap")
	if mission == nil && bootstrap && persona.CopyResearchMap.AudienceSummary == "" && len(persona.CopyResearchMap.SuggestedTags) == 0 {
		updateProgress("產生拷貝忍者研究地圖…", 8)
		if err := ensureCopyResearchMap(ctx, deps, tenantID, ownerUID, persona, memberCtx, updateProgress); err != nil {
			return err
		}
		persona, err = deps.Persona.Get(ctx, tenantID, ownerUID, personaID)
		if err != nil {
			return err
		}
	}

	keywords := stringSliceField(payload, "keywords")
	exclusions := persona.CopyResearchMap.Exclusions
	missionScan := mission != nil
	if missionScan {
		if len(keywords) == 0 {
			keywords = append([]string(nil), mission.SelectedTags...)
		}
		exclusions = mission.ResearchMap.Exclusions
		if len(keywords) == 0 {
			return fmt.Errorf("請先產生研究地圖並勾選搜尋標籤")
		}
	} else if len(keywords) == 0 {
		keywords = deriveViralKeywords(persona)
	}
	if len(keywords) == 0 {
		return fmt.Errorf("請提供爆款掃描關鍵字，或先完成研究地圖／對標帳號")
	}

	if missionScan && missionID != "" {
		updateProgress("清除舊爆款與產文草稿…", 8)
		if deps.ScanPost != nil {
			if err := deps.ScanPost.ClearCopyMissionViralScan(ctx, tenantID, ownerUID, personaID, missionID); err != nil {
				return err
			}
		}
		if deps.CopyDraft != nil {
			if err := deps.CopyDraft.ClearByMission(ctx, tenantID, ownerUID, personaID, missionID); err != nil {
				return err
			}
		}
	}

	updateProgress("準備爆款掃描…", 12)
	crawlerFn := makeCrawlerSearchFn(ScanPlacementDeps{ThreadsAccount: deps.ThreadsAccount}, tenantID, ownerUID)
	candidates, err := libviral.RunDiscover(ctx, libviral.DiscoverInput{
		Keywords:    keywords,
		Exclusions:  exclusions,
		SeedQuery:   missionSeedQuery(mission),
		Label:       missionLabel(mission),
		TopicHints:  missionTopicHints(mission),
		Member:      memberCtx,
		Crawler:     crawlerFn,
		MissionScan: missionScan,
	}, updateProgress)
	if err != nil {
		return err
	}

	if missionScan && memberCtx.ScrapeReplies && memberCtx.ApiConnected {
		updateProgress("收集高互動留言樣本…", 88)
		candidates = placement.AttachReplies(ctx, placement.ScrapeRepliesInput{
			Posts:          candidates,
			Member:         memberCtx,
			RepliesPerPost: memberCtx.RepliesPerPost,
			MaxPosts:       6,
		})
	}

	updateProgress(fmt.Sprintf("寫入 %d 篇爆款候選…", len(candidates)), 92)
	count, err := deps.ScanPost.ReplaceFromViralScan(ctx, scanpostusecase.ViralReplaceRequest{
		TenantID:      tenantID,
		OwnerUID:      ownerUID,
		PersonaID:     personaID,
		CopyMissionID: missionID,
		ScanJobID:     step.JobID,
		Posts:         candidates,
	})
	if err != nil {
		return err
	}

	if missionScan && missionID != "" && mission != nil {
		scanned := missionentity.StatusScanned
		jobID := step.JobID
		prevSimilar := similarAccountsFromSummary(mission.ResearchMap.SimilarAccounts)
		var knownAccounts map[string]accountentity.KnownAccountProfile
		if memberCtx.ActiveAccountID != "" && deps.ThreadsAccount != nil {
			knownAccounts, _ = deps.ThreadsAccount.GetKnownAccounts(ctx, tenantID, ownerUID, memberCtx.ActiveAccountID)
		}
		excluded := libviral.ExcludedSimilarAccountUsernames(prevSimilar)
		excluded = append(excluded, libviral.ExcludedKnownAccountUsernames(knownAccounts)...)
		referenceAccounts := libviral.BuildReferenceAccountsFromScan(libviral.ReferenceAccountInput{
			SeedQuery:         mission.SeedQuery,
			Label:             mission.Label,
			Posts:             candidates,
			Limit:             libviral.MaxSimilarAccounts,
			ExcludedUsernames: excluded,
		})
		// Web-search fallback: only when scan surfaced fewer than
		// MinSimilarAccountsBeforeWebSupplement (5) reference authors — avoids
		// extra search API calls when scan data is already sufficient.
		// (site:threads.net "seed" + pillars). Failures are logged-quiet and must
		// not fail the scan job — the scan-derived accounts are still valid.
		if len(referenceAccounts) < libviral.MinSimilarAccountsBeforeWebSupplement && placement.WebSearchAvailable(research) {
			referenceAccounts = supplementSimilarAccountsFromWeb(ctx, referenceAccounts, memberCtx, mission)
		}
		entityTags := make([]missionentity.SuggestedTag, 0, len(mission.ResearchMap.SuggestedTags))
		for _, tag := range mission.ResearchMap.SuggestedTags {
			entityTags = append(entityTags, missionentity.SuggestedTag{
				Tag:          tag.Tag,
				Reason:       tag.Reason,
				SearchIntent: tag.SearchIntent,
				SearchType:   tag.SearchType,
			})
		}
		mergedSimilar := libviral.MergeSimilarAccounts(prevSimilar, referenceAccounts)
		mergedSimilar = libviral.ApplyKnownAccountMemory(mergedSimilar, knownAccounts)
		if memberCtx.ActiveAccountID != "" && deps.ThreadsAccount != nil {
			_ = deps.ThreadsAccount.UpsertKnownAccountsFromScan(ctx, threadsaccountdomain.UpsertKnownAccountsFromScanRequest{
				TenantID:        tenantID,
				OwnerUID:        ownerUID,
				AccountID:       memberCtx.ActiveAccountID,
				MissionID:       missionID,
				PersonaID:       personaID,
				SelectedTags:    mission.SelectedTags,
				SimilarAccounts: mergedSimilar,
			})
		}
		var audienceSamples []missionentity.AudienceSample
		if memberCtx.ApiConnected && memberCtx.ScrapeReplies {
			var allReplies []placement.ReplyCandidate
			for _, post := range candidates {
				allReplies = append(allReplies, post.Replies...)
			}
			if len(allReplies) > 0 {
				audienceSamples = libviral.MergeAudienceSamples(
					audienceSamplesFromSummary(mission.ResearchMap.AudienceSamples),
					libviral.BuildAudienceSamplesFromReplies(allReplies, libviral.AudienceOpts{
						Max:            libviral.MaxAudienceSamples,
						ExcludeAuthors: libviral.SimilarAccountUsernames(mergedSimilar),
						Now:            clock.NowUnixNano(),
					}),
				)
			}
		}
		updatedMap := missionentity.ResearchMap{
			AudienceSummary: mission.ResearchMap.AudienceSummary,
			ContentGoal:     mission.ResearchMap.ContentGoal,
			Questions:       append([]string(nil), mission.ResearchMap.Questions...),
			Pillars:         append([]string(nil), mission.ResearchMap.Pillars...),
			Exclusions:      append([]string(nil), mission.ResearchMap.Exclusions...),
			SuggestedTags:   entityTags,
			SimilarAccounts: mergedSimilar,
			AudienceSamples: audienceSamples,
			BenchmarkNotes:  mission.ResearchMap.BenchmarkNotes,
		}
		_, _ = deps.CopyMission.Update(ctx, missiondomain.UpdateRequest{
			TenantID:  tenantID,
			OwnerUID:  ownerUID,
			PersonaID: personaID,
			MissionID: missionID,
			Patch: missiondomain.MissionPatch{
				LastScanJobID: &jobID,
				Status:        &scanned,
				ResearchMap:   &updatedMap,
			},
		})
	}

	nextRoute := "/matrix"
	if missionID != "" {
		nextRoute = fmt.Sprintf("/matrix/missions/%s", missionID)
	}
	handoff := map[string]any{
		"flow":            "copy",
		"persona_id":      personaID,
		"copy_mission_id": missionID,
		"summary":         fmt.Sprintf("爆款掃描完成：%d 篇候選", count),
		"next_route":      nextRoute,
	}

	_, err = deps.Jobs.CompleteRun(ctx, jobdom.CompleteRunRequest{
		JobID:    step.JobID,
		WorkerID: step.WorkerID,
		Result: map[string]any{
			"post_count": count,
			"handoff":    handoff,
		},
	})
	return err
}

func ensureCopyResearchMap(
	ctx context.Context,
	deps ScanViralDeps,
	tenantID, ownerUID string,
	persona *personadomain.PersonaSummary,
	memberCtx placement.MemberContext,
	updateProgress func(string, int),
) error {
	if persona == nil {
		return fmt.Errorf("copy research map: missing persona")
	}
	credential, err := deps.ThreadsAccount.ResolveMemberAiCredential(ctx, tenantID, ownerUID)
	if err != nil {
		return err
	}
	providerID, err := aiusecase.MapWorkerProvider(credential.Provider)
	if err != nil {
		return err
	}

	label := strings.TrimSpace(persona.DisplayName)
	if label == "" {
		label = "拷貝主題"
	}
	seed := strings.TrimSpace(persona.SeedQuery)
	if seed == "" {
		seed = strings.TrimPrefix(strings.TrimSpace(persona.StyleBenchmark), "@")
	}
	if seed == "" {
		for _, line := range strings.Split(strings.TrimSpace(persona.Brief), "\n") {
			line = strings.TrimSpace(line)
			if line != "" {
				seed = line
				break
			}
		}
	}

	result, err := deps.AI.GenerateText(ctx, domai.GenerateRequest{
		Provider: providerID,
		Model:    credential.Model,
		Credential: domai.Credential{
			APIKey: credential.APIKey,
		},
		System: libviral.BuildCopyResearchMapSystemPrompt(),
		Messages: []domai.Message{
			{
				Role: "user",
				Content: libviral.BuildCopyResearchMapUserPrompt(libviral.CopyResearchMapInput{
					Label:          label,
					SeedQuery:      seed,
					Brief:          persona.Brief,
					Persona:        persona.Persona,
					StyleBenchmark: persona.StyleBenchmark,
				}),
			},
		},
	})
	if err != nil {
		return err
	}

	parsed, err := libviral.ParseCopyResearchMapOutput(result.Text)
	if err != nil {
		return app.For(code.AI).SvcThirdParty("拷貝研究地圖 LLM 回傳無法解析：" + err.Error())
	}

	entityMap := personaentity.CopyResearchMap{
		AudienceSummary: parsed.AudienceSummary,
		ContentGoal:     parsed.ContentGoal,
		Questions:       parsed.Questions,
		Pillars:         parsed.Pillars,
		Exclusions:      parsed.Exclusions,
		SuggestedTags:   parsed.SuggestedTags,
		BenchmarkNotes:  parsed.BenchmarkNotes,
	}
	patch := personadomain.PersonaPatch{
		CopyResearchMap: &entityMap,
	}
	if seed != "" && strings.TrimSpace(persona.SeedQuery) == "" {
		patch.SeedQuery = &seed
	}
	_, err = deps.Persona.Update(ctx, personadomain.UpdateRequest{
		TenantID:  tenantID,
		OwnerUID:  ownerUID,
		PersonaID: persona.ID,
		Patch:     patch,
	})
	if err != nil {
		return err
	}
	if updateProgress != nil {
		updateProgress("研究地圖已就緒", 10)
	}
	_ = memberCtx
	return nil
}

func personaIDFromPayload(payload map[string]any) string {
	if id := stringField(payload, "persona_id"); id != "" {
		return id
	}
	return stringField(payload, "scope_id")
}

func missionSeedQuery(mission *missiondomain.MissionSummary) string {
	if mission == nil {
		return ""
	}
	return mission.SeedQuery
}

func missionLabel(mission *missiondomain.MissionSummary) string {
	if mission == nil {
		return ""
	}
	return mission.Label
}

func missionTopicHints(mission *missiondomain.MissionSummary) []string {
	if mission == nil {
		return nil
	}
	out := []string{}
	seen := map[string]struct{}{}
	add := func(item string) {
		item = strings.TrimSpace(item)
		if item == "" {
			return
		}
		if _, ok := seen[item]; ok {
			return
		}
		seen[item] = struct{}{}
		out = append(out, item)
	}
	for _, tag := range mission.SelectedTags {
		add(tag)
	}
	for _, tag := range mission.ResearchMap.SuggestedTags {
		add(tag.Tag)
	}
	for _, pillar := range mission.ResearchMap.Pillars {
		add(pillar)
	}
	for _, question := range mission.ResearchMap.Questions {
		add(question)
	}
	return out
}

func deriveViralKeywords(persona *personadomain.PersonaSummary) []string {
	if persona == nil {
		return nil
	}
	out := []string{}
	seen := map[string]struct{}{}
	add := func(kw string) {
		kw = strings.TrimSpace(kw)
		if kw == "" {
			return
		}
		if _, ok := seen[kw]; ok {
			return
		}
		seen[kw] = struct{}{}
		out = append(out, kw)
	}
	for _, tag := range persona.CopyResearchMap.SuggestedTags {
		add(tag)
	}
	for _, q := range persona.CopyResearchMap.Questions {
		add(q)
	}
	if bench := strings.TrimPrefix(strings.TrimSpace(persona.StyleBenchmark), "@"); bench != "" {
		add(bench)
	}
	for _, line := range strings.Split(strings.TrimSpace(persona.Brief), "\n") {
		line = strings.TrimSpace(line)
		if line != "" {
			add(line)
			break
		}
	}
	return out
}

// supplementSimilarAccountsFromWeb asks the configured web search provider
// for threads.net profiles matching the mission seed/brief/pillars, then
// merges the results with the scan-derived accounts via EnrichSimilarAccounts.
// It must never return an error: web search outage is non-fatal. Confidence is
// upgraded to "high" for usernames that appear in both sources (scan+web).
func supplementSimilarAccountsFromWeb(
	ctx context.Context,
	scanAccounts []missionentity.SimilarAccount,
	memberCtx placement.MemberContext,
	mission *missiondomain.MissionSummary,
) []missionentity.SimilarAccount {
	if memberCtx.WebSearchAPIKey() == "" {
		return scanAccounts
	}
	client := websearch.New(memberCtx.WebSearchConfig())
	if !client.Enabled() {
		return scanAccounts
	}
	webAccounts, err := libviral.DiscoverSimilarAccounts(ctx, client, libviral.DiscoverAccountsInput{
		SeedQuery: mission.SeedQuery,
		Brief:     mission.Brief,
		Pillars:   mission.ResearchMap.Pillars,
	})
	if err != nil || len(webAccounts) == 0 {
		return scanAccounts
	}
	entityWeb := make([]missionentity.SimilarAccount, 0, len(webAccounts))
	for _, w := range webAccounts {
		matched := w.MatchedSource
		if len(matched) == 0 {
			matched = []string{"web"}
		}
		entityWeb = append(entityWeb, missionentity.SimilarAccount{
			Username:      w.Username,
			Reason:        w.Reason,
			Source:        "web",
			MatchedSource: matched,
			Confidence:    w.Confidence,
			ProfileURL:    w.ProfileURL,
		})
	}
	return libviral.MergeSimilarAccounts(scanAccounts, entityWeb)
}