tm_back/cmd/worker/workers/summarizer.go

package workers

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

// DocumentSummarizationWorker triggers AI summarization for scraped tenders.
// The AI service owns MinIO artefacts; this worker marks Mongo processing flags
// after the pipeline or on-demand summarize API completes.
type DocumentSummarizationWorker struct {
	Mongo      *mongo.ConnectionManager
	Logger     logger.Logger
	TenderRepo tender.TenderRepository
	AIClient   *ai_summarizer.Client
	AIStorage  *ai_summarizer.StorageClient
}

// NewDocumentSummarizationWorker creates a document summarization worker.
func NewDocumentSummarizationWorker(
	mongo *mongo.ConnectionManager,
	logger logger.Logger,
	tenderRepo tender.TenderRepository,
	aiClient *ai_summarizer.Client,
	aiStorage *ai_summarizer.StorageClient,
) *DocumentSummarizationWorker {
	return &DocumentSummarizationWorker{
		Mongo:      mongo,
		Logger:     logger,
		TenderRepo: tenderRepo,
		AIClient:   aiClient,
		AIStorage:  aiStorage,
	}
}

// Run starts the document summarization process.
func (w *DocumentSummarizationWorker) Run() {
	w.Logger.Info("Document summarization worker started", map[string]interface{}{})

	if w.AIClient == nil {
		w.Logger.Warn("AI summarizer client not configured, skipping document summarization run", map[string]interface{}{})
		return
	}

	if refs := w.collectUnSummarizedTenderRefs(context.Background()); len(refs) > 0 {
		if _, err := w.AIClient.TriggerSummarizeBatch(context.Background(), refs); err != nil {
			w.Logger.Warn("Failed to trigger AI summarize batch (continuing with per-tender sync)", map[string]interface{}{
				"tender_count": len(refs),
				"error":        err.Error(),
			})
		}
	}

	limit := 5
	skip := 0
	for {
		tenders, totalCount, err := w.TenderRepo.GetUnSummarizedTenders(context.Background(), limit, skip)
		if err != nil {
			w.Logger.Error("Failed to get un-summarized tenders", map[string]interface{}{
				"error": err.Error(),
			})
			return
		}

		w.Logger.Info("Found tenders for document summarization", map[string]interface{}{
			"count":       len(tenders),
			"total_count": totalCount,
			"skip":        skip,
		})

		if len(tenders) == 0 {
			break
		}

		for i := range tenders {
			t := &tenders[i]
			w.Logger.Info("Processing tender for document summarization", map[string]interface{}{
				"tender_id":             t.ID.Hex(),
				"notice_publication_id": t.NoticePublicationID,
			})
			w.summarizeTender(t)
		}

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

	w.Logger.Info("Document summarization worker completed", map[string]interface{}{})
}

func (w *DocumentSummarizationWorker) collectUnSummarizedTenderRefs(ctx context.Context) []ai_summarizer.TenderRef {
	limit := 100
	skip := 0
	refs := make([]ai_summarizer.TenderRef, 0)

	for {
		tenders, totalCount, err := w.TenderRepo.GetUnSummarizedTenders(ctx, limit, skip)
		if err != nil {
			w.Logger.Warn("Failed to collect tenders for summarize batch", map[string]interface{}{
				"error": err.Error(),
			})
			return refs
		}
		if len(tenders) == 0 {
			return refs
		}

		for i := range tenders {
			t := &tenders[i]
			if strings.TrimSpace(t.ContractFolderID) == "" || strings.TrimSpace(t.NoticePublicationID) == "" {
				continue
			}
			refs = append(refs, ai_summarizer.NewTenderRef(t.ContractFolderID, t.NoticePublicationID))
		}

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

func (w *DocumentSummarizationWorker) summarizeTender(t *tender.Tender) {
	if strings.TrimSpace(t.NoticePublicationID) == "" {
		w.Logger.Warn("Skipping tender summarization: missing notice_publication_id", map[string]interface{}{
			"tender_id": t.ID.Hex(),
		})
		w.markTenderSummarized(t)
		return
	}
	if strings.TrimSpace(t.ContractFolderID) == "" {
		w.Logger.Warn("Skipping tender summarization: missing contract_folder_id", map[string]interface{}{
			"tender_id": t.ID.Hex(),
		})
		w.markTenderSummarized(t)
		return
	}

	ctx := context.Background()
	if w.isSummaryReadyInStorage(ctx, t) {
		w.markTenderSummarized(t)
		return
	}

	req := ai_summarizer.NewSummarizeRequest(t.ContractFolderID, t.NoticePublicationID)
	if _, err := w.AIClient.FetchSummaryOnDemand(ctx, req); err != nil {
		w.Logger.Error("Failed to summarize tender via AI service", map[string]interface{}{
			"tender_id":             t.ID.Hex(),
			"notice_publication_id": t.NoticePublicationID,
			"contract_folder_id":    t.ContractFolderID,
			"error":                 err.Error(),
		})
		return
	}

	if !w.isSummaryReadyInStorage(ctx, t) {
		w.Logger.Warn("AI summarize completed but storage is not ready yet", map[string]interface{}{
			"tender_id":             t.ID.Hex(),
			"notice_publication_id": t.NoticePublicationID,
		})
	}

	w.markTenderSummarized(t)
}

func (w *DocumentSummarizationWorker) isSummaryReadyInStorage(ctx context.Context, t *tender.Tender) bool {
	if w.AIStorage == nil {
		return false
	}
	ready, err := w.AIStorage.IsSummaryReady(ctx, t.ContractFolderID, t.NoticePublicationID)
	if err != nil {
		w.Logger.Debug("Could not check summarization status in storage", map[string]interface{}{
			"tender_id":             t.ID.Hex(),
			"notice_publication_id": t.NoticePublicationID,
			"error":                 err.Error(),
		})
		return false
	}
	return ready
}

func (w *DocumentSummarizationWorker) markTenderSummarized(t *tender.Tender) {
	now := time.Now().Unix()
	t.ProcessingMetadata.DocumentsSummarized = true
	t.ProcessingMetadata.DocumentsSummarizedAt = now
	t.UpdatedAt = now

	if err := w.TenderRepo.Update(context.Background(), t); err != nil {
		w.Logger.Error("Failed to update tender summarization status", map[string]interface{}{
			"tender_id":             t.ID.Hex(),
			"notice_publication_id": t.NoticePublicationID,
			"error":                 err.Error(),
		})
		return
	}

	w.Logger.Info("Tender marked as summarized", map[string]interface{}{
		"tender_id":             t.ID.Hex(),
		"notice_publication_id": t.NoticePublicationID,
	})
}