package company import ( "context" "encoding/base64" "encoding/json" "errors" "fmt" "io" "strings" "time" "github.com/redis/go-redis/v9" "golang.org/x/sync/errgroup" "tm/pkg/ai_summarizer" ) var ErrAIRecommendationNotConfigured = errors.New("AI recommendation service is not configured") const aiRecommendationCacheKeyPrefix = "ai:recommendations:" const ( recommendationFetchMaxAttempts = 6 recommendationFetchInitialDelay = 10 * time.Second ) func normalizeRecommendationCompanyIDs(companyIDs []string) []string { normalized := make([]string, 0, len(companyIDs)) seen := make(map[string]struct{}, len(companyIDs)) for _, companyID := range companyIDs { clean := strings.TrimSpace(companyID) if clean == "" { continue } if _, ok := seen[clean]; ok { continue } seen[clean] = struct{}{} normalized = append(normalized, clean) } return normalized } func mergeRecommendedTenders(recommendationLists ...[]RecommendedTenderResponse) []RecommendedTenderResponse { merged := make([]RecommendedTenderResponse, 0) seen := make(map[string]struct{}) for _, recommendations := range recommendationLists { for _, recommendation := range recommendations { tenderID := strings.TrimSpace(recommendation.TenderID) if tenderID == "" { continue } if _, ok := seen[tenderID]; ok { continue } seen[tenderID] = struct{}{} recommendation.TenderID = tenderID recommendation.Analysis = strings.TrimSpace(recommendation.Analysis) merged = append(merged, recommendation) } } return merged } func aiRecommendationCacheKey(companyID string) string { return aiRecommendationCacheKeyPrefix + companyID } // AIRecommendationClient defines the interface for company tender recommendation AI operations. type AIRecommendationClient interface { StartOnboarding(ctx context.Context, reqBody ai_summarizer.OnboardingRequest) (*ai_summarizer.OnboardingResponse, error) FetchRecommendations(ctx context.Context, reqBody ai_summarizer.RecommendRequest) ([]ai_summarizer.RecommendedTender, error) } type onboardingDocumentPayload struct { Filename string `json:"filename"` ContentType string `json:"content_type"` Content string `json:"content"` } // StartAIOnboarding sends company profile data to the AI team to start recommendation onboarding. func (s *companyService) StartAIOnboarding(ctx context.Context, companyID string) (*OnboardingResponse, error) { if s.aiRecommendationClient == nil { return nil, ErrAIRecommendationNotConfigured } companyID = strings.TrimSpace(companyID) if companyID == "" { return nil, errors.New("company ID is required") } s.invalidateAIRecommendationCache(ctx, companyID) result, err := s.startAIOnboardingInternal(ctx, companyID) if err != nil { return nil, err } s.scheduleRecommendationRefreshAfterOnboarding(companyID) return result, nil } // triggerAIOnboardingAsync re-indexes the company on the AI service and refreshes the recommendation cache. // This is the only path (besides StartAIOnboarding) that calls the AI recommend endpoint. func (s *companyService) triggerAIOnboardingAsync(companyID string) { if s.aiRecommendationClient == nil { return } companyID = strings.TrimSpace(companyID) if companyID == "" { return } go func() { ctx := context.Background() s.logger.Info("Starting async AI company onboarding and recommendation refresh", map[string]interface{}{ "company_id": companyID, }) s.invalidateAIRecommendationCache(ctx, companyID) if _, err := s.startAIOnboardingInternal(ctx, companyID); err != nil { s.logger.Error("Async AI company onboarding failed", map[string]interface{}{ "company_id": companyID, "error": err.Error(), }) return } if err := s.fetchAndCacheAIRecommendationsWithRetry(ctx, companyID); err != nil { s.logger.Error("Failed to refresh AI recommendations after onboarding", map[string]interface{}{ "company_id": companyID, "error": err.Error(), }) return } s.logger.Info("Async AI company onboarding and recommendation refresh completed", map[string]interface{}{ "company_id": companyID, }) }() } func (s *companyService) startAIOnboardingInternal(ctx context.Context, companyID string) (*OnboardingResponse, error) { company, err := s.repository.GetByID(ctx, companyID) if err != nil { s.logger.Error("Failed to load company for AI onboarding", map[string]interface{}{ "company_id": companyID, "error": err.Error(), }) return nil, errors.New("company not found") } documents, err := s.buildOnboardingDocuments(company.DocumentFileIDs) if err != nil { s.logger.Error("Failed to prepare company documents for AI onboarding", map[string]interface{}{ "company_id": companyID, "error": err.Error(), }) return nil, fmt.Errorf("failed to prepare company documents: %w", err) } websiteURL := "" if company.Website != nil { websiteURL = strings.TrimSpace(*company.Website) } s.logger.Info("Starting AI company onboarding", map[string]interface{}{ "company_id": companyID, "documents_count": len(company.DocumentFileIDs), "links_count": len(company.Links), "has_website_url": websiteURL != "", }) result, err := s.aiRecommendationClient.StartOnboarding(ctx, ai_summarizer.OnboardingRequest{ CompanyID: companyID, Documents: documents, WebsiteURL: websiteURL, Links: companyLinkURLs(company.Links), }) if err != nil { s.logger.Error("AI onboarding request failed", map[string]interface{}{ "company_id": companyID, "error": err.Error(), }) return nil, fmt.Errorf("failed to start AI onboarding: %w", err) } return &OnboardingResponse{Status: result.Status}, nil } // GetAIRecommendations returns cached recommendations only. The AI service is not called on read; // cache is populated when a company is created, updated, or onboarded. func (s *companyService) GetAIRecommendations(ctx context.Context, companyID string) ([]RecommendedTenderResponse, error) { if s.aiRecommendationClient == nil { return nil, ErrAIRecommendationNotConfigured } companyID = strings.TrimSpace(companyID) if companyID == "" { return nil, errors.New("company ID is required") } if _, err := s.repository.GetByID(ctx, companyID); err != nil { s.logger.Error("Failed to load company for AI recommendations", map[string]interface{}{ "company_id": companyID, "error": err.Error(), }) return nil, errors.New("company not found") } if cached, ok := s.getCachedAIRecommendations(ctx, companyID); ok { return cached, nil } s.logger.Info("AI recommendations cache miss; returning empty list", map[string]interface{}{ "company_id": companyID, }) return []RecommendedTenderResponse{}, nil } func (s *companyService) validateRecommendationCompanyIDs(ctx context.Context, companyIDs []string) error { companies, err := s.repository.GetByIDs(ctx, companyIDs) if err != nil { s.logger.Error("Failed to load companies for merged AI recommendations", map[string]interface{}{ "company_ids": companyIDs, "error": err.Error(), }) return errors.New("company not found") } if len(companies) != len(companyIDs) { s.logger.Error("Some companies were not found for merged AI recommendations", map[string]interface{}{ "requested_count": len(companyIDs), "loaded_count": len(companies), }) return errors.New("company not found") } return nil } func (s *companyService) getCachedAIRecommendationsOrEmpty(ctx context.Context, companyID string) []RecommendedTenderResponse { if cached, ok := s.getCachedAIRecommendations(ctx, companyID); ok { return cached } s.logger.Info("AI recommendations cache miss; returning empty list", map[string]interface{}{ "company_id": companyID, }) return []RecommendedTenderResponse{} } // GetMergedAIRecommendations unions cached AI recommendations across multiple companies. func (s *companyService) GetMergedAIRecommendations(ctx context.Context, companyIDs []string) ([]RecommendedTenderResponse, error) { if s.aiRecommendationClient == nil { return nil, ErrAIRecommendationNotConfigured } normalized := normalizeRecommendationCompanyIDs(companyIDs) if len(normalized) == 0 { return []RecommendedTenderResponse{}, nil } if err := s.validateRecommendationCompanyIDs(ctx, normalized); err != nil { return nil, err } recommendationLists := make([][]RecommendedTenderResponse, len(normalized)) group, groupCtx := errgroup.WithContext(ctx) for i, companyID := range normalized { i := i companyID := companyID group.Go(func() error { recommendationLists[i] = s.getCachedAIRecommendationsOrEmpty(groupCtx, companyID) return nil }) } if err := group.Wait(); err != nil { return nil, err } return mergeRecommendedTenders(recommendationLists...), nil } func (s *companyService) scheduleRecommendationRefreshAfterOnboarding(companyID string) { if s.aiRecommendationClient == nil { return } companyID = strings.TrimSpace(companyID) if companyID == "" { return } go func() { ctx := context.Background() if err := s.fetchAndCacheAIRecommendationsWithRetry(ctx, companyID); err != nil { s.logger.Error("Failed to refresh AI recommendation cache after onboarding", map[string]interface{}{ "company_id": companyID, "error": err.Error(), }) } }() } // fetchAndCacheAIRecommendationsWithRetry waits for the AI service to finish indexing after onboarding. // Empty results are not cached so reads keep returning empty until real recommendations exist. func (s *companyService) fetchAndCacheAIRecommendationsWithRetry(ctx context.Context, companyID string) error { delay := recommendationFetchInitialDelay for attempt := 1; attempt <= recommendationFetchMaxAttempts; attempt++ { if attempt > 1 { s.logger.Info("Retrying AI recommendation fetch after onboarding", map[string]interface{}{ "company_id": companyID, "attempt": attempt, "delay_sec": int(delay.Seconds()), }) } else { s.logger.Info("Waiting before first AI recommendation fetch after onboarding", map[string]interface{}{ "company_id": companyID, "delay_sec": int(delay.Seconds()), }) } select { case <-ctx.Done(): return ctx.Err() case <-time.After(delay): } if attempt > 1 { delay *= 2 } responses, err := s.fetchAIRecommendations(ctx, companyID) if err != nil { if attempt == recommendationFetchMaxAttempts { return err } continue } if len(responses) == 0 { if attempt == recommendationFetchMaxAttempts { s.logger.Warn("AI returned no recommendations after onboarding retries; cache not updated", map[string]interface{}{ "company_id": companyID, "attempts": attempt, }) return nil } continue } s.cacheAIRecommendations(ctx, companyID, responses) s.logger.Info("AI recommendation cache populated after onboarding", map[string]interface{}{ "company_id": companyID, "count": len(responses), "attempt": attempt, }) return nil } return nil } func (s *companyService) fetchAIRecommendations(ctx context.Context, companyID string) ([]RecommendedTenderResponse, error) { s.logger.Info("Fetching AI tender recommendations from AI service", map[string]interface{}{ "company_id": companyID, }) items, err := s.aiRecommendationClient.FetchRecommendations(ctx, ai_summarizer.RecommendRequest{ CompanyID: companyID, }) if err != nil { s.logger.Error("AI recommend request failed", map[string]interface{}{ "company_id": companyID, "error": err.Error(), }) return nil, fmt.Errorf("failed to fetch AI recommendations: %w", err) } responses := make([]RecommendedTenderResponse, 0, len(items)) for _, item := range items { responses = append(responses, RecommendedTenderResponse{ Rank: item.Rank, TenderID: item.TenderID, Analysis: item.Analysis, }) } return responses, nil } func (s *companyService) getCachedAIRecommendations(ctx context.Context, companyID string) ([]RecommendedTenderResponse, bool) { if s.redisClient == nil { return nil, false } raw, err := s.redisClient.Get(ctx, aiRecommendationCacheKey(companyID)) if err != nil { if err != redis.Nil { s.logger.Warn("Failed to read AI recommendation cache", map[string]interface{}{ "company_id": companyID, "error": err.Error(), }) } return nil, false } var responses []RecommendedTenderResponse if err := json.Unmarshal([]byte(raw), &responses); err != nil { s.logger.Warn("Failed to decode AI recommendation cache", map[string]interface{}{ "company_id": companyID, "error": err.Error(), }) _ = s.redisClient.Del(ctx, aiRecommendationCacheKey(companyID)) return nil, false } if len(responses) == 0 { _ = s.redisClient.Del(ctx, aiRecommendationCacheKey(companyID)) return nil, false } s.logger.Debug("AI tender recommendations served from cache", map[string]interface{}{ "company_id": companyID, "count": len(responses), }) return responses, true } func (s *companyService) cacheAIRecommendations(ctx context.Context, companyID string, responses []RecommendedTenderResponse) { if s.redisClient == nil || len(responses) == 0 { return } encoded, err := json.Marshal(responses) if err != nil { s.logger.Warn("Failed to encode AI recommendations for cache", map[string]interface{}{ "company_id": companyID, "error": err.Error(), }) return } expiration := s.recommendationCacheExpiration() if err := s.redisClient.Set(ctx, aiRecommendationCacheKey(companyID), string(encoded), expiration); err != nil { s.logger.Warn("Failed to store AI recommendation cache", map[string]interface{}{ "company_id": companyID, "error": err.Error(), }) } } func (s *companyService) recommendationCacheExpiration() time.Duration { if s.recommendationCacheTTL > 0 { return s.recommendationCacheTTL } return 0 } func (s *companyService) invalidateAIRecommendationCache(ctx context.Context, companyID string) { if s.redisClient == nil { return } if err := s.redisClient.Del(ctx, aiRecommendationCacheKey(companyID)); err != nil { s.logger.Warn("Failed to invalidate AI recommendation cache", map[string]interface{}{ "company_id": companyID, "error": err.Error(), }) } } func (s *companyService) buildOnboardingDocuments(fileIDs []string) (string, error) { if len(fileIDs) == 0 { return "[]", nil } if s.fileStore == nil { return "", errors.New("file storage is not configured") } payloads := make([]onboardingDocumentPayload, 0, len(fileIDs)) for _, fileID := range fileIDs { if strings.TrimSpace(fileID) == "" { continue } reader, info, err := s.fileStore.Download(fileID) if err != nil { s.logger.Warn("Skipping company document for AI onboarding", map[string]interface{}{ "file_id": fileID, "error": err.Error(), }) continue } content, readErr := io.ReadAll(reader) closeErr := reader.Close() if readErr != nil { return "", fmt.Errorf("failed to read document %s: %w", fileID, readErr) } if closeErr != nil { s.logger.Warn("Failed to close company document reader", map[string]interface{}{ "file_id": fileID, "error": closeErr.Error(), }) } filename := fileID contentType := "application/octet-stream" if info != nil { if info.Filename != "" { filename = info.Filename } if info.ContentType != "" { contentType = info.ContentType } } payloads = append(payloads, onboardingDocumentPayload{ Filename: filename, ContentType: contentType, Content: base64.StdEncoding.EncodeToString(content), }) } encoded, err := json.Marshal(payloads) if err != nil { return "", fmt.Errorf("failed to encode company documents: %w", err) } return string(encoded), nil }