tm_back/internal/company/recommendation_pipeline_refresh.go

package company

import (
	"context"
	"strings"
	"sync/atomic"
	"time"

	"golang.org/x/sync/errgroup"

	"tm/pkg/ai_summarizer"
)

// AIPipelineStatusClient reports pipeline job status from the AI service.
type AIPipelineStatusClient interface {
	GetPipelineLastRun(ctx context.Context) (*ai_summarizer.PipelineReportResponse, error)
}

const (
	recommendationPipelineWaitInitialDelay = 5 * time.Minute
	recommendationPipelineWaitMaxAttempts  = 24
	recommendationRefreshConcurrency       = 3
)

// ScheduleRefreshCachedAIRecommendationsAfterPipeline waits for the AI daily pipeline to finish,
// then re-fetches ranked tenders for every company that already has a recommendation cache.
func (s *companyService) ScheduleRefreshCachedAIRecommendationsAfterPipeline() {
	if s.aiRecommendationClient == nil {
		return
	}

	go func() {
		ctx := context.Background()
		s.logger.Info("Scheduling AI recommendation cache refresh after pipeline sync", map[string]interface{}{})

		if err := s.refreshCachedAIRecommendationsAfterPipeline(ctx); err != nil {
			s.logger.Error("AI recommendation cache refresh after pipeline failed", map[string]interface{}{
				"error": err.Error(),
			})
			return
		}

		s.logger.Info("AI recommendation cache refresh after pipeline completed", map[string]interface{}{})
	}()
}

func (s *companyService) refreshCachedAIRecommendationsAfterPipeline(ctx context.Context) error {
	if err := s.waitForPipelineDailyRunCompletion(ctx); err != nil {
		return err
	}

	companyIDs, err := s.listCompanyIDsWithRecommendationCache(ctx)
	if err != nil {
		return err
	}
	if len(companyIDs) == 0 {
		s.logger.Info("No companies with recommendation cache to refresh after pipeline", map[string]interface{}{})
		return nil
	}

	s.logger.Info("Refreshing AI recommendation caches after pipeline", map[string]interface{}{
		"company_count": len(companyIDs),
	})

	group, groupCtx := errgroup.WithContext(ctx)
	group.SetLimit(recommendationRefreshConcurrency)

	var refreshed atomic.Int32
	var failed atomic.Int32
	for _, companyID := range companyIDs {
		companyID := companyID
		group.Go(func() error {
			if err := s.refreshAIRecommendationsCache(groupCtx, companyID); err != nil {
				s.logger.Error("Failed to refresh AI recommendation cache for company", map[string]interface{}{
					"company_id": companyID,
					"error":      err.Error(),
				})
				failed.Add(1)
				return nil
			}
			refreshed.Add(1)
			return nil
		})
	}

	if err := group.Wait(); err != nil {
		return err
	}

	s.logger.Info("AI recommendation cache refresh summary", map[string]interface{}{
		"company_count": len(companyIDs),
		"refreshed":     refreshed.Load(),
		"failed":        failed.Load(),
	})
	return nil
}

func (s *companyService) waitForPipelineDailyRunCompletion(ctx context.Context) error {
	if s.aiPipelineStatusClient == nil {
		s.logger.Info("AI pipeline status client not configured; waiting before recommendation refresh", map[string]interface{}{
			"delay_sec": int(recommendationPipelineWaitInitialDelay.Seconds()),
		})
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(recommendationPipelineWaitInitialDelay):
		}
		return nil
	}

	delay := recommendationPipelineWaitInitialDelay
	for attempt := 1; attempt <= recommendationPipelineWaitMaxAttempts; attempt++ {
		if attempt == 1 {
			s.logger.Info("Waiting before first AI pipeline status check", map[string]interface{}{
				"delay_sec": int(delay.Seconds()),
			})
		} else {
			s.logger.Info("Retrying AI pipeline status check", map[string]interface{}{
				"attempt":   attempt,
				"delay_sec": int(delay.Seconds()),
			})
		}

		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(delay):
		}

		if attempt > 1 {
			delay *= 2
			if delay > 30*time.Minute {
				delay = 30 * time.Minute
			}
		}

		report, err := s.aiPipelineStatusClient.GetPipelineLastRun(ctx)
		if err != nil {
			s.logger.Warn("Failed to read AI pipeline last-run status", map[string]interface{}{
				"attempt": attempt,
				"error":   err.Error(),
			})
			if attempt == recommendationPipelineWaitMaxAttempts {
				s.logger.Warn("Proceeding with recommendation refresh without pipeline status", map[string]interface{}{})
				return nil
			}
			continue
		}

		if !isPipelineDailyRunInProgress(report) {
			s.logger.Info("AI pipeline daily-run finished; refreshing recommendation caches", map[string]interface{}{
				"attempt": attempt,
				"status":  strings.TrimSpace(report.Status),
			})
			return nil
		}

		if attempt == recommendationPipelineWaitMaxAttempts {
			s.logger.Warn("AI pipeline still in progress after max wait; refreshing recommendation caches anyway", map[string]interface{}{
				"attempts": attempt,
				"status":   strings.TrimSpace(report.Status),
			})
			return nil
		}
	}

	return nil
}

func isPipelineDailyRunInProgress(report *ai_summarizer.PipelineReportResponse) bool {
	if report == nil {
		return true
	}
	status := strings.ToLower(strings.TrimSpace(report.Status))
	switch status {
	case "", "running", "in_progress", "started", "processing", "pending", "queued":
		return true
	default:
		return false
	}
}

func (s *companyService) listCompanyIDsWithRecommendationCache(ctx context.Context) ([]string, error) {
	allIDs, err := s.repository.ListIDs(ctx)
	if err != nil {
		s.logger.Error("Failed to list companies for recommendation refresh", map[string]interface{}{
			"error": err.Error(),
		})
		return nil, err
	}

	cached := make([]string, 0, len(allIDs))
	for _, companyID := range allIDs {
		if _, ok := s.getCachedAIRecommendations(ctx, companyID); ok {
			cached = append(cached, companyID)
		}
	}
	return cached, nil
}

// refreshAIRecommendationsCache updates Redis when the AI service returns a non-empty ranked list.
// Empty or failed fetches leave the existing cache untouched.
func (s *companyService) refreshAIRecommendationsCache(ctx context.Context, companyID string) error {
	responses, err := s.fetchAIRecommendations(ctx, companyID)
	if err != nil {
		return err
	}
	if len(responses) == 0 {
		s.logger.Info("Skipping recommendation cache update after pipeline: AI returned empty list", map[string]interface{}{
			"company_id": companyID,
		})
		return nil
	}

	s.cacheAIRecommendations(ctx, companyID, responses)
	s.logger.Info("AI recommendation cache refreshed after pipeline", map[string]interface{}{
		"company_id": companyID,
		"count":      len(responses),
	})
	return nil
}