package workers import ( "context" "strings" "tm/internal/tender" "tm/pkg/ai_summarizer" "tm/pkg/logger" "tm/pkg/mongo" ) // TranslationWorker syncs translation completion from the AI pipeline MinIO storage. // Batch translation is triggered via POST /ai/translate/batch; this worker only // verifies which tenders now have translations available in storage. 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 triggers the AI translation pipeline and logs tenders with ready translations in MinIO. 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 refs := w.collectUnTranslatedTenderRefs(context.Background()); len(refs) > 0 { if _, err := w.AIClient.TriggerTranslateBatch(context.Background(), refs, w.TargetLanguages); err != nil { w.Logger.Warn("Failed to trigger AI translate batch", map[string]interface{}{ "tender_count": len(refs), "target_languages": w.TargetLanguages, "error": err.Error(), }) } } readyCount := 0 for _, language := range w.TargetLanguages { readyCount += w.syncLanguageFromStorage(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, "translations_ready_in_storage": readyCount, "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) syncLanguageFromStorage(language string) int { if w.AIStorage == nil { w.Logger.Warn("Translation storage client not configured; skipping MinIO sync", map[string]interface{}{ "target_language": language, }) return 0 } readyCount := 0 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 sync", map[string]interface{}{ "target_language": language, "error": err.Error(), }) return readyCount } w.Logger.Info("Fetched tenders for translation sync", map[string]interface{}{ "target_language": language, "batch_count": len(tenders), "total_count": totalCount, "skip": skip, }) if len(tenders) == 0 { return readyCount } for _, t := range tenders { if w.hasTranslationInStorage(&t, language) { readyCount++ w.Logger.Debug("Translation available in MinIO", map[string]interface{}{ "tender_id": t.ID.Hex(), "notice_id": t.NoticePublicationID, "target_language": language, }) } } skip += len(tenders) if int64(skip) >= totalCount { return readyCount } } } func (w *TranslationWorker) hasTranslationInStorage(t *tender.Tender, language string) bool { if strings.TrimSpace(t.ContractFolderID) == "" || t.NoticePublicationID == "" { return false } stored, err := w.AIStorage.GetTranslationFromStorage( context.Background(), t.ContractFolderID, t.NoticePublicationID, language, ) return err == nil && strings.TrimSpace(stored.Title) != "" } func (w *TranslationWorker) collectUnTranslatedTenderRefs(ctx context.Context) []ai_summarizer.TenderRef { seen := make(map[string]struct{}) refs := make([]ai_summarizer.TenderRef, 0) for _, language := range w.TargetLanguages { skip := 0 for { tenders, totalCount, err := w.TenderRepo.GetUnTranslatedTenders(ctx, language, w.BatchSize, skip) if err != nil { w.Logger.Warn("Failed to collect tenders for translate batch", map[string]interface{}{ "target_language": language, "error": err.Error(), }) break } if len(tenders) == 0 { break } for _, t := range tenders { if strings.TrimSpace(t.ContractFolderID) == "" || strings.TrimSpace(t.NoticePublicationID) == "" { continue } key := t.ContractFolderID + "|" + t.NoticePublicationID if _, dup := seen[key]; dup { continue } seen[key] = struct{}{} refs = append(refs, ai_summarizer.NewTenderRef(t.ContractFolderID, t.NoticePublicationID)) } skip += len(tenders) if int64(skip) >= totalCount { break } } } return refs }