Mazyar
1edf42187d
continuous-integration/drone/push Build is passing
- Added functionality to retrieve merged AI recommendations for multiple companies, improving the relevance of tender suggestions based on company-specific data. - Introduced normalization functions to clean and deduplicate company IDs, ensuring accurate processing of recommendations. - Enhanced the company context resolution in customer middleware to support multiple assigned companies, improving the handling of company-specific requests. - Updated the tender recommendation logic to utilize the new merged recommendations and handle exclusions for rejected tenders accordingly. - Added unit tests to verify the new recommendation merging logic and company ID normalization, ensuring robust functionality. This update significantly enhances the tender recommendation process by allowing for more comprehensive and relevant suggestions based on multiple company contexts, improving user experience and satisfaction.
497 lines
14 KiB
Go
497 lines
14 KiB
Go
package company
|
|
|
|
import (
|
|
"context"
|
|
"encoding/base64"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/redis/go-redis/v9"
|
|
|
|
"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),
|
|
"has_website_url": websiteURL != "",
|
|
})
|
|
|
|
result, err := s.aiRecommendationClient.StartOnboarding(ctx, ai_summarizer.OnboardingRequest{
|
|
CompanyID: companyID,
|
|
Documents: documents,
|
|
WebsiteURL: websiteURL,
|
|
})
|
|
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
|
|
}
|
|
|
|
// 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
|
|
}
|
|
|
|
recommendationLists := make([][]RecommendedTenderResponse, 0, len(normalized))
|
|
for _, companyID := range normalized {
|
|
recommendations, err := s.GetAIRecommendations(ctx, companyID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
recommendationLists = append(recommendationLists, recommendations)
|
|
}
|
|
|
|
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
|
|
}
|