tm_back/internal/ai_pipeline/service.go

package ai_pipeline

import (
	"context"
	"fmt"

	"tm/pkg/ai_summarizer"
	"tm/pkg/logger"
)

// Client defines Opplens AI operations exposed to the admin API.
type Client interface {
	ScrapeDocuments(ctx context.Context, reqBody ai_summarizer.ScrapeDocumentsRequest) (*ai_summarizer.ScrapeDocumentsResponse, error)
	ScrapeDocumentsBatch(ctx context.Context, tenders []ai_summarizer.TenderRef) (*ai_summarizer.BatchAcceptedResponse, error)
	GetScrapePortals(ctx context.Context) (*ai_summarizer.ScrapePortalsResponse, error)
	TriggerSummarizeBatch(ctx context.Context, tenders []ai_summarizer.TenderRef) (*ai_summarizer.BatchAcceptedResponse, error)
	TriggerTranslateBatch(ctx context.Context, tenders []ai_summarizer.TenderRef, languages []string) (*ai_summarizer.BatchAcceptedResponse, error)
	PipelineSync(ctx context.Context) (*ai_summarizer.PipelineSyncResponse, error)
	PipelineRun(ctx context.Context, reqBody ai_summarizer.PipelineRunRequest) (map[string]interface{}, error)
	PipelineRunBatch(ctx context.Context, tenders []ai_summarizer.TenderRef, languages []string) (*ai_summarizer.BatchAcceptedResponse, error)
	PipelineAnalyze(ctx context.Context) (*ai_summarizer.PipelineActionResponse, error)
	PipelineDailyRun(ctx context.Context) (*ai_summarizer.PipelineStartedResponse, error)
	PipelineAuto(ctx context.Context) (*ai_summarizer.PipelineStartedResponse, error)
	GetPipelineLastRun(ctx context.Context) (*ai_summarizer.PipelineReportResponse, error)
	GetPipelineLastAutoRun(ctx context.Context) (*ai_summarizer.PipelineReportResponse, error)
	GetPipelineReport(ctx context.Context, window string) (*ai_summarizer.PipelineReportResponse, error)
	GetPipelineStats(ctx context.Context, window string) (*ai_summarizer.PipelineStatsResponse, error)
	GetPipelineMinioStats(ctx context.Context) (ai_summarizer.PipelineMinioStatsResponse, error)
}

// Service proxies admin pipeline operations to the Opplens AI service.
type Service interface {
	ScrapeDocuments(ctx context.Context, form ScrapeDocumentsForm) (*ai_summarizer.ScrapeDocumentsResponse, error)
	ScrapeDocumentsBatch(ctx context.Context, form TenderBatchForm) (*ai_summarizer.BatchAcceptedResponse, error)
	GetScrapePortals(ctx context.Context) (*ai_summarizer.ScrapePortalsResponse, error)
	SummarizeBatch(ctx context.Context, form TenderBatchForm) (*ai_summarizer.BatchAcceptedResponse, error)
	TranslateBatch(ctx context.Context, form TranslateBatchForm) (*ai_summarizer.BatchAcceptedResponse, error)
	Sync(ctx context.Context) (*ai_summarizer.PipelineSyncResponse, error)
	Run(ctx context.Context, form PipelineRunForm) (map[string]interface{}, error)
	RunBatch(ctx context.Context, form PipelineRunBatchForm) (*ai_summarizer.BatchAcceptedResponse, error)
	Analyze(ctx context.Context) (*ai_summarizer.PipelineActionResponse, error)
	DailyRun(ctx context.Context) (*ai_summarizer.PipelineStartedResponse, error)
	Auto(ctx context.Context) (*ai_summarizer.PipelineStartedResponse, error)
	GetLastRun(ctx context.Context) (*ai_summarizer.PipelineReportResponse, error)
	GetLastAutoRun(ctx context.Context) (*ai_summarizer.PipelineReportResponse, error)
	GetReport(ctx context.Context, window string) (*ai_summarizer.PipelineReportResponse, error)
	GetStats(ctx context.Context, window string) (*ai_summarizer.PipelineStatsResponse, error)
	GetMinioStats(ctx context.Context) (ai_summarizer.PipelineMinioStatsResponse, error)
}

// CachedRecommendationRefresher schedules a background refresh of company recommendation caches.
type CachedRecommendationRefresher interface {
	ScheduleRefreshCachedAIRecommendationsAfterPipeline()
}

// ScrapedDocumentMetadataSyncer persists MinIO scraped-document metadata onto tender records.
type ScrapedDocumentMetadataSyncer interface {
	SyncScrapedDocumentsFromStorage(ctx context.Context, contractFolderID, noticePublicationID string) error
}

type service struct {
	client                  Client
	logger                  logger.Logger
	metadataSyncer          ScrapedDocumentMetadataSyncer
	recommendationRefresher CachedRecommendationRefresher
}

// NewService creates an AI pipeline admin service.
func NewService(
	client Client,
	log logger.Logger,
	metadataSyncer ScrapedDocumentMetadataSyncer,
	recommendationRefresher CachedRecommendationRefresher,
) Service {
	return &service{
		client:                  client,
		logger:                  log,
		metadataSyncer:          metadataSyncer,
		recommendationRefresher: recommendationRefresher,
	}
}

func (s *service) requireClient() error {
	if s.client == nil {
		return ErrAINotConfigured
	}
	return nil
}

func (s *service) ScrapeDocuments(ctx context.Context, form ScrapeDocumentsForm) (*ai_summarizer.ScrapeDocumentsResponse, error) {
	if err := s.requireClient(); err != nil {
		return nil, err
	}

	s.logger.Info("Admin scrape documents requested", map[string]interface{}{
		"contract_folder_id":    form.ContractFolderID,
		"notice_publication_id": form.NoticePublicationID,
	})

	resp, err := s.client.ScrapeDocuments(ctx, ai_summarizer.ScrapeDocumentsRequest{
		ContractFolderID:    form.ContractFolderID,
		NoticePublicationID: form.NoticePublicationID,
	})
	if err != nil {
		s.logger.Error("Admin scrape documents failed", map[string]interface{}{
			"contract_folder_id":    form.ContractFolderID,
			"notice_publication_id": form.NoticePublicationID,
			"error":                 err.Error(),
		})
		return nil, fmt.Errorf("scrape documents failed: %w", err)
	}

	if resp != nil && resp.Success && resp.FilesDownloaded > 0 {
		s.syncScrapedDocumentMetadata(ctx, form.ContractFolderID, form.NoticePublicationID)
	}

	return resp, nil
}

func (s *service) ScrapeDocumentsBatch(ctx context.Context, form TenderBatchForm) (*ai_summarizer.BatchAcceptedResponse, error) {
	if err := s.requireClient(); err != nil {
		return nil, err
	}
	refs, err := validateTenderBatch(form.Tenders)
	if err != nil {
		return nil, err
	}

	s.logger.Info("Admin scrape documents batch requested", map[string]interface{}{
		"tender_count": len(refs),
	})

	resp, err := s.client.ScrapeDocumentsBatch(ctx, refs)
	if err != nil {
		s.logger.Error("Admin scrape documents batch failed", map[string]interface{}{
			"tender_count": len(refs),
			"error":        err.Error(),
		})
		return nil, fmt.Errorf("scrape documents batch failed: %w", err)
	}

	return resp, nil
}

func (s *service) GetScrapePortals(ctx context.Context) (*ai_summarizer.ScrapePortalsResponse, error) {
	if err := s.requireClient(); err != nil {
		return nil, err
	}

	s.logger.Info("Admin get scrape portals requested", map[string]interface{}{})

	resp, err := s.client.GetScrapePortals(ctx)
	if err != nil {
		s.logger.Error("Admin get scrape portals failed", map[string]interface{}{
			"error": err.Error(),
		})
		return nil, fmt.Errorf("get scrape portals failed: %w", err)
	}

	return resp, nil
}

func (s *service) SummarizeBatch(ctx context.Context, form TenderBatchForm) (*ai_summarizer.BatchAcceptedResponse, error) {
	if err := s.requireClient(); err != nil {
		return nil, err
	}
	refs, err := validateTenderBatch(form.Tenders)
	if err != nil {
		return nil, err
	}

	s.logger.Info("Admin summarize batch requested", map[string]interface{}{
		"tender_count": len(refs),
	})

	resp, err := s.client.TriggerSummarizeBatch(ctx, refs)
	if err != nil {
		s.logger.Error("Admin summarize batch failed", map[string]interface{}{
			"tender_count": len(refs),
			"error":        err.Error(),
		})
		return nil, fmt.Errorf("summarize batch failed: %w", err)
	}

	return resp, nil
}

func (s *service) TranslateBatch(ctx context.Context, form TranslateBatchForm) (*ai_summarizer.BatchAcceptedResponse, error) {
	if err := s.requireClient(); err != nil {
		return nil, err
	}
	refs, err := validateTenderBatch(form.Tenders)
	if err != nil {
		return nil, err
	}
	languages := normalizeLanguages(form.Languages)
	if len(languages) == 0 {
		return nil, ErrEmptyLanguages
	}

	s.logger.Info("Admin translate batch requested", map[string]interface{}{
		"tender_count": len(refs),
		"languages":    languages,
	})

	resp, err := s.client.TriggerTranslateBatch(ctx, refs, languages)
	if err != nil {
		s.logger.Error("Admin translate batch failed", map[string]interface{}{
			"tender_count": len(refs),
			"languages":    languages,
			"error":        err.Error(),
		})
		return nil, fmt.Errorf("translate batch failed: %w", err)
	}

	return resp, nil
}

func (s *service) Sync(ctx context.Context) (*ai_summarizer.PipelineSyncResponse, error) {
	if err := s.requireClient(); err != nil {
		return nil, err
	}

	s.logger.Info("Admin pipeline sync requested", map[string]interface{}{})

	resp, err := s.client.PipelineSync(ctx)
	if err != nil {
		s.logger.Error("Admin pipeline sync failed", map[string]interface{}{
			"error": err.Error(),
		})
		return nil, fmt.Errorf("pipeline sync failed: %w", err)
	}

	s.scheduleRecommendationRefreshAfterPipeline()

	return resp, nil
}

func (s *service) Run(ctx context.Context, form PipelineRunForm) (map[string]interface{}, error) {
	if err := s.requireClient(); err != nil {
		return nil, err
	}
	languages := normalizeLanguages(form.Languages)
	if len(languages) == 0 {
		return nil, ErrEmptyLanguages
	}

	req := ai_summarizer.PipelineRunRequest{
		ContractFolderID:    form.ContractFolderID,
		NoticePublicationID: form.NoticePublicationID,
		Languages:           languages,
	}

	s.logger.Info("Admin pipeline run requested", map[string]interface{}{
		"contract_folder_id":    req.ContractFolderID,
		"notice_publication_id": req.NoticePublicationID,
		"languages":             languages,
	})

	resp, err := s.client.PipelineRun(ctx, req)
	if err != nil {
		s.logger.Error("Admin pipeline run failed", map[string]interface{}{
			"contract_folder_id":    req.ContractFolderID,
			"notice_publication_id": req.NoticePublicationID,
			"error":                 err.Error(),
		})
		return nil, fmt.Errorf("pipeline run failed: %w", err)
	}

	s.syncScrapedDocumentMetadata(ctx, req.ContractFolderID, req.NoticePublicationID)

	return resp, nil
}

func (s *service) RunBatch(ctx context.Context, form PipelineRunBatchForm) (*ai_summarizer.BatchAcceptedResponse, error) {
	if err := s.requireClient(); err != nil {
		return nil, err
	}
	refs, err := validateTenderBatch(form.Tenders)
	if err != nil {
		return nil, err
	}
	languages := normalizeLanguages(form.Languages)
	if len(languages) == 0 {
		return nil, ErrEmptyLanguages
	}

	s.logger.Info("Admin pipeline run batch requested", map[string]interface{}{
		"tender_count": len(refs),
		"languages":    languages,
	})

	resp, err := s.client.PipelineRunBatch(ctx, refs, languages)
	if err != nil {
		s.logger.Error("Admin pipeline run batch failed", map[string]interface{}{
			"tender_count": len(refs),
			"languages":    languages,
			"error":        err.Error(),
		})
		return nil, fmt.Errorf("pipeline run batch failed: %w", err)
	}

	return resp, nil
}

func (s *service) Analyze(ctx context.Context) (*ai_summarizer.PipelineActionResponse, error) {
	if err := s.requireClient(); err != nil {
		return nil, err
	}

	s.logger.Info("Admin pipeline analyze requested", map[string]interface{}{})

	resp, err := s.client.PipelineAnalyze(ctx)
	if err != nil {
		s.logger.Error("Admin pipeline analyze failed", map[string]interface{}{
			"error": err.Error(),
		})
		return nil, fmt.Errorf("pipeline analyze failed: %w", err)
	}

	return resp, nil
}

func (s *service) DailyRun(ctx context.Context) (*ai_summarizer.PipelineStartedResponse, error) {
	if err := s.requireClient(); err != nil {
		return nil, err
	}

	s.logger.Info("Admin pipeline daily-run requested", map[string]interface{}{})

	resp, err := s.client.PipelineDailyRun(ctx)
	if err != nil {
		s.logger.Error("Admin pipeline daily-run failed", map[string]interface{}{
			"error": err.Error(),
		})
		return nil, fmt.Errorf("pipeline daily-run failed: %w", err)
	}

	s.scheduleRecommendationRefreshAfterPipeline()

	return resp, nil
}

func (s *service) Auto(ctx context.Context) (*ai_summarizer.PipelineStartedResponse, error) {
	if err := s.requireClient(); err != nil {
		return nil, err
	}

	s.logger.Info("Admin pipeline auto requested", map[string]interface{}{})

	resp, err := s.client.PipelineAuto(ctx)
	if err != nil {
		s.logger.Error("Admin pipeline auto failed", map[string]interface{}{
			"error": err.Error(),
		})
		return nil, fmt.Errorf("pipeline auto failed: %w", err)
	}

	s.scheduleRecommendationRefreshAfterPipeline()

	return resp, nil
}

func (s *service) GetLastRun(ctx context.Context) (*ai_summarizer.PipelineReportResponse, error) {
	if err := s.requireClient(); err != nil {
		return nil, err
	}

	resp, err := s.client.GetPipelineLastRun(ctx)
	if err != nil {
		s.logger.Error("Admin get pipeline last-run failed", map[string]interface{}{
			"error": err.Error(),
		})
		return nil, fmt.Errorf("get pipeline last-run failed: %w", err)
	}

	return resp, nil
}

func (s *service) GetLastAutoRun(ctx context.Context) (*ai_summarizer.PipelineReportResponse, error) {
	if err := s.requireClient(); err != nil {
		return nil, err
	}

	resp, err := s.client.GetPipelineLastAutoRun(ctx)
	if err != nil {
		s.logger.Error("Admin get pipeline last-auto-run failed", map[string]interface{}{
			"error": err.Error(),
		})
		return nil, fmt.Errorf("get pipeline last-auto-run failed: %w", err)
	}

	return resp, nil
}

func (s *service) GetReport(ctx context.Context, window string) (*ai_summarizer.PipelineReportResponse, error) {
	if err := s.requireClient(); err != nil {
		return nil, err
	}

	resp, err := s.client.GetPipelineReport(ctx, window)
	if err != nil {
		s.logger.Error("Admin get pipeline report failed", map[string]interface{}{
			"window": window,
			"error":  err.Error(),
		})
		return nil, fmt.Errorf("get pipeline report failed: %w", err)
	}

	return resp, nil
}

func (s *service) GetStats(ctx context.Context, window string) (*ai_summarizer.PipelineStatsResponse, error) {
	if err := s.requireClient(); err != nil {
		return nil, err
	}

	resp, err := s.client.GetPipelineStats(ctx, window)
	if err != nil {
		s.logger.Error("Admin get pipeline stats failed", map[string]interface{}{
			"window": window,
			"error":  err.Error(),
		})
		return nil, fmt.Errorf("get pipeline stats failed: %w", err)
	}

	return resp, nil
}

func (s *service) GetMinioStats(ctx context.Context) (ai_summarizer.PipelineMinioStatsResponse, error) {
	if err := s.requireClient(); err != nil {
		return nil, err
	}

	resp, err := s.client.GetPipelineMinioStats(ctx)
	if err != nil {
		s.logger.Error("Admin get pipeline minio stats failed", map[string]interface{}{
			"error": err.Error(),
		})
		return nil, fmt.Errorf("get pipeline minio stats failed: %w", err)
	}

	return resp, nil
}

func (s *service) scheduleRecommendationRefreshAfterPipeline() {
	if s.recommendationRefresher == nil {
		return
	}
	s.recommendationRefresher.ScheduleRefreshCachedAIRecommendationsAfterPipeline()
}

func validateTenderBatch(forms []TenderRefForm) ([]ai_summarizer.TenderRef, error) {
	if len(forms) == 0 {
		return nil, ErrEmptyTenderBatch
	}
	return toTenderRefs(forms), nil
}

func (s *service) syncScrapedDocumentMetadata(ctx context.Context, contractFolderID, noticePublicationID string) {
	if s.metadataSyncer == nil {
		return
	}
	if err := s.metadataSyncer.SyncScrapedDocumentsFromStorage(ctx, contractFolderID, noticePublicationID); err != nil {
		s.logger.Warn("Failed to sync scraped document metadata to Mongo", map[string]interface{}{
			"contract_folder_id":    contractFolderID,
			"notice_publication_id": noticePublicationID,
			"error":                 err.Error(),
		})
	}
}