tm_back/cmd/worker/workers/translation.go

package workers

import (
	"context"
	"errors"
	"strings"
	"tm/internal/tender"
	"tm/pkg/ai_summarizer"
	"tm/pkg/logger"
	"tm/pkg/mongo"
)

// TranslationWorker backfills missing tender translations using the AI pipeline
// and on-demand translate API. Results are persisted by the AI service in MinIO only.
type TranslationWorker struct {
	Mongo                    *mongo.ConnectionManager
	Logger                   logger.Logger
	TenderRepo               tender.TenderRepository
	AIClient                 *ai_summarizer.Client
	AIStorage                *ai_summarizer.StorageClient
	TranslationSuccessCounter *mongo.Counter
	TargetLanguages          []string
	BatchSize                int
}

// NewTranslationWorker creates a translation worker for the given target languages.
func NewTranslationWorker(
	mongo *mongo.ConnectionManager,
	logger logger.Logger,
	tenderRepo tender.TenderRepository,
	aiClient *ai_summarizer.Client,
	aiStorage *ai_summarizer.StorageClient,
	translationSuccessCounter *mongo.Counter,
	targetLanguages []string,
	batchSize int,
) *TranslationWorker {
	langs := make([]string, 0, len(targetLanguages))
	seen := make(map[string]struct{}, len(targetLanguages))
	for _, lang := range targetLanguages {
		clean := strings.ToLower(strings.TrimSpace(lang))
		if clean == "" {
			continue
		}
		if _, dup := seen[clean]; dup {
			continue
		}
		seen[clean] = struct{}{}
		langs = append(langs, clean)
	}
	if len(langs) == 0 {
		langs = []string{"en"}
	}
	if batchSize <= 0 {
		batchSize = 20
	}

	return &TranslationWorker{
		Mongo:                     mongo,
		Logger:                    logger,
		TenderRepo:                tenderRepo,
		AIClient:                  aiClient,
		AIStorage:                 aiStorage,
		TranslationSuccessCounter: translationSuccessCounter,
		TargetLanguages:           langs,
		BatchSize:                 batchSize,
	}
}

// Run starts backfilling translations for tenders missing target-language content.
func (w *TranslationWorker) Run() {
	if w.AIClient == nil {
		w.Logger.Warn("Translation worker skipped: AI client is not configured", map[string]interface{}{
			"target_languages": w.TargetLanguages,
		})
		return
	}

	w.Logger.Info("Translation worker started", map[string]interface{}{
		"target_languages": w.TargetLanguages,
		"batch_size":       w.BatchSize,
	})

	startCount := w.getSuccessfulTranslationCount(mongo.AITranslationSuccessCounterKeyDailyJob)

	if _, err := w.AIClient.TriggerPipelineTranslate(context.Background(), w.TargetLanguages); err != nil {
		w.Logger.Warn("Failed to trigger AI pipeline translate (continuing with per-tender sync)", map[string]interface{}{
			"target_languages": w.TargetLanguages,
			"error":            err.Error(),
		})
	}

	for _, language := range w.TargetLanguages {
		w.runForLanguage(language)
	}

	endCount := w.getSuccessfulTranslationCount(mongo.AITranslationSuccessCounterKeyDailyJob)
	totalCount := w.getSuccessfulTranslationCount(mongo.AITranslationSuccessCounterKey)
	runCount := endCount - startCount
	if runCount < 0 {
		runCount = 0
	}

	w.Logger.Info("Translation worker completed", map[string]interface{}{
		"target_languages":                               w.TargetLanguages,
		"successful_ai_translation_requests_run":         runCount,
		"successful_ai_translation_requests_daily_job":   endCount,
		"successful_ai_translation_requests_total":       totalCount,
	})
}

func (w *TranslationWorker) getSuccessfulTranslationCount(key string) int64 {
	if w.TranslationSuccessCounter == nil {
		return 0
	}
	count, err := w.TranslationSuccessCounter.Get(context.Background(), key)
	if err != nil {
		w.Logger.Warn("Failed to read AI translation success counter", map[string]interface{}{
			"counter_key": key,
			"error":       err.Error(),
		})
		return 0
	}
	return count
}

func (w *TranslationWorker) runForLanguage(language string) {
	skip := 0
	for {
		tenders, totalCount, err := w.TenderRepo.GetUnTranslatedTenders(context.Background(), language, w.BatchSize, skip)
		if err != nil {
			w.Logger.Error("Failed to fetch tenders for translation backfill", map[string]interface{}{
				"target_language": language,
				"error":           err.Error(),
			})
			return
		}

		w.Logger.Info("Fetched tenders for translation backfill", map[string]interface{}{
			"target_language": language,
			"batch_count":     len(tenders),
			"total_count":     totalCount,
			"skip":            skip,
		})

		if len(tenders) == 0 {
			return
		}

		for _, t := range tenders {
			w.translateTender(&t, language)
		}

		skip += len(tenders)
		if int64(skip) >= totalCount {
			return
		}
	}
}

func (w *TranslationWorker) translateTender(t *tender.Tender, language string) {
	if t.NoticePublicationID == "" {
		return
	}
	if strings.TrimSpace(t.ContractFolderID) == "" {
		w.Logger.Debug("Skipping tender translation: missing contract_folder_id", map[string]interface{}{
			"tender_id": t.ID.Hex(),
		})
		return
	}
	if w.hasTranslationInStorage(t, language) {
		w.Logger.Debug("Skipping tender translation: already available in MinIO", map[string]interface{}{
			"tender_id":       t.ID.Hex(),
			"notice_id":       t.NoticePublicationID,
			"target_language": language,
		})
		return
	}

	title, description, source, err := w.resolveTranslation(t, language)
	if err != nil {
		w.Logger.Error("Failed to translate tender", map[string]interface{}{
			"tender_id":          t.ID.Hex(),
			"notice_id":          t.NoticePublicationID,
			"target_language":    language,
			"translation_source": source,
			"translation_error":  err.Error(),
		})
		return
	}

	w.Logger.Info("Tender translated successfully", map[string]interface{}{
		"tender_id":          t.ID.Hex(),
		"notice_id":          t.NoticePublicationID,
		"target_language":    language,
		"translation_source": source,
		"title_len":          len(title),
		"description_len":    len(description),
	})
}

func (w *TranslationWorker) hasTranslationInStorage(t *tender.Tender, language string) bool {
	if w.AIStorage == nil {
		return false
	}
	stored, err := w.AIStorage.GetTranslationFromStorage(
		context.Background(),
		t.ContractFolderID,
		t.NoticePublicationID,
		language,
	)
	return err == nil && strings.TrimSpace(stored.Title) != ""
}

func (w *TranslationWorker) resolveTranslation(t *tender.Tender, language string) (title, description, source string, err error) {
	if w.AIStorage != nil {
		stored, storageErr := w.AIStorage.GetTranslationFromStorage(
			context.Background(),
			t.ContractFolderID,
			t.NoticePublicationID,
			language,
		)
		if storageErr == nil {
			return stored.Title, stored.Description, "storage", nil
		}
		if !errors.Is(storageErr, ai_summarizer.ErrTranslationNotReady) &&
			!errors.Is(storageErr, ai_summarizer.ErrObjectNotFound) &&
			!errors.Is(storageErr, ai_summarizer.ErrNoTranslation) {
			w.Logger.Debug("Translation not available from storage yet", map[string]interface{}{
				"tender_id":       t.ID.Hex(),
				"notice_id":       t.NoticePublicationID,
				"target_language": language,
				"error":           storageErr.Error(),
			})
		}
	}

	resp, apiErr := w.AIClient.FetchTranslationOnDemand(context.Background(), ai_summarizer.TranslateRequest{
		ContractFolderID:    t.ContractFolderID,
		NoticePublicationID: t.NoticePublicationID,
		Title:               t.Title,
		Description:         t.Description,
		Language:            language,
		RequestSource:       ai_summarizer.TranslationRequestSourceDailyJob,
	})
	if apiErr != nil {
		return "", "", "api", apiErr
	}
	return resp.TranslatedTitle, resp.TranslatedDescription, "api", nil
}