Enhance AI recommendation caching and onboarding process in company service
continuous-integration/drone/push Build is passing
continuous-integration/drone/push Build is passing
- Updated `AISummarizerConfig` to allow for a default `RecommendationCacheTTL` of 0, enabling persistent caching until company updates. - Refactored `StartAIOnboarding` to include cache invalidation and asynchronous recommendation refresh, improving responsiveness during onboarding. - Introduced `triggerAIOnboardingAsync` method for background processing of AI onboarding and cache refresh, enhancing user experience. - Improved logging for AI onboarding and recommendation fetching processes, providing better observability and error tracking. This update optimizes the AI recommendation caching mechanism and onboarding workflow, ensuring a smoother and more efficient experience for users.
This commit is contained in:
@@ -69,8 +69,9 @@ type AISummarizerConfig struct {
|
|||||||
APIRetryCount int `env:"AI_SUMMARIZER_API_RETRY_COUNT" envDefault:"2"`
|
APIRetryCount int `env:"AI_SUMMARIZER_API_RETRY_COUNT" envDefault:"2"`
|
||||||
APIRetryDelay time.Duration `env:"AI_SUMMARIZER_API_RETRY_DELAY" envDefault:"3s"`
|
APIRetryDelay time.Duration `env:"AI_SUMMARIZER_API_RETRY_DELAY" envDefault:"3s"`
|
||||||
DefaultLanguage string `env:"AI_SUMMARIZER_DEFAULT_LANGUAGE" envDefault:"en"`
|
DefaultLanguage string `env:"AI_SUMMARIZER_DEFAULT_LANGUAGE" envDefault:"en"`
|
||||||
// RecommendationCacheTTL caches AI /recommend responses per company. Set to 0 to disable.
|
// RecommendationCacheTTL is an optional max Redis TTL for cached recommendations.
|
||||||
RecommendationCacheTTL time.Duration `env:"AI_RECOMMENDATION_CACHE_TTL" envDefault:"15m"`
|
// When 0 (default), entries persist until the company is created, updated, or onboarded.
|
||||||
|
RecommendationCacheTTL time.Duration `env:"AI_RECOMMENDATION_CACHE_TTL" envDefault:"0"`
|
||||||
|
|
||||||
// MinIO storage settings
|
// MinIO storage settings
|
||||||
MinioEndpoint string `env:"AI_SUMMARIZER_MINIO_ENDPOINT" envDefault:""`
|
MinioEndpoint string `env:"AI_SUMMARIZER_MINIO_ENDPOINT" envDefault:""`
|
||||||
|
|||||||
@@ -8,6 +8,7 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
"strings"
|
"strings"
|
||||||
|
"time"
|
||||||
|
|
||||||
"github.com/redis/go-redis/v9"
|
"github.com/redis/go-redis/v9"
|
||||||
|
|
||||||
@@ -39,10 +40,67 @@ func (s *companyService) StartAIOnboarding(ctx context.Context, companyID string
|
|||||||
if s.aiRecommendationClient == nil {
|
if s.aiRecommendationClient == nil {
|
||||||
return nil, ErrAIRecommendationNotConfigured
|
return nil, ErrAIRecommendationNotConfigured
|
||||||
}
|
}
|
||||||
if strings.TrimSpace(companyID) == "" {
|
|
||||||
|
companyID = strings.TrimSpace(companyID)
|
||||||
|
if companyID == "" {
|
||||||
return nil, errors.New("company ID is required")
|
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.refreshAIRecommendationsCacheAsync(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.fetchAndCacheAIRecommendations(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)
|
company, err := s.repository.GetByID(ctx, companyID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
s.logger.Error("Failed to load company for AI onboarding", map[string]interface{}{
|
s.logger.Error("Failed to load company for AI onboarding", map[string]interface{}{
|
||||||
@@ -67,9 +125,9 @@ func (s *companyService) StartAIOnboarding(ctx context.Context, companyID string
|
|||||||
}
|
}
|
||||||
|
|
||||||
s.logger.Info("Starting AI company onboarding", map[string]interface{}{
|
s.logger.Info("Starting AI company onboarding", map[string]interface{}{
|
||||||
"company_id": companyID,
|
"company_id": companyID,
|
||||||
"documents_count": len(company.DocumentFileIDs),
|
"documents_count": len(company.DocumentFileIDs),
|
||||||
"has_website_url": websiteURL != "",
|
"has_website_url": websiteURL != "",
|
||||||
})
|
})
|
||||||
|
|
||||||
result, err := s.aiRecommendationClient.StartOnboarding(ctx, ai_summarizer.OnboardingRequest{
|
result, err := s.aiRecommendationClient.StartOnboarding(ctx, ai_summarizer.OnboardingRequest{
|
||||||
@@ -85,57 +143,21 @@ func (s *companyService) StartAIOnboarding(ctx context.Context, companyID string
|
|||||||
return nil, fmt.Errorf("failed to start AI onboarding: %w", err)
|
return nil, fmt.Errorf("failed to start AI onboarding: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Prefetch recommendations in the background; keep serving the previous cache until refresh completes.
|
|
||||||
s.refreshAIRecommendationsCacheAsync(companyID)
|
|
||||||
|
|
||||||
return &OnboardingResponse{Status: result.Status}, nil
|
return &OnboardingResponse{Status: result.Status}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// triggerAIOnboardingAsync starts AI onboarding in the background after company profile changes.
|
// GetAIRecommendations returns cached recommendations only. The AI service is not called on read;
|
||||||
func (s *companyService) triggerAIOnboardingAsync(companyID string) {
|
// cache is populated when a company is created, updated, or onboarded.
|
||||||
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", map[string]interface{}{
|
|
||||||
"company_id": companyID,
|
|
||||||
})
|
|
||||||
|
|
||||||
if _, err := s.StartAIOnboarding(ctx, companyID); err != nil {
|
|
||||||
s.logger.Error("Async AI company onboarding failed", map[string]interface{}{
|
|
||||||
"company_id": companyID,
|
|
||||||
"error": err.Error(),
|
|
||||||
})
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
s.logger.Info("Async AI company onboarding completed", map[string]interface{}{
|
|
||||||
"company_id": companyID,
|
|
||||||
})
|
|
||||||
}()
|
|
||||||
}
|
|
||||||
|
|
||||||
// GetAIRecommendations retrieves ranked tender recommendations from the AI team.
|
|
||||||
func (s *companyService) GetAIRecommendations(ctx context.Context, companyID string) ([]RecommendedTenderResponse, error) {
|
func (s *companyService) GetAIRecommendations(ctx context.Context, companyID string) ([]RecommendedTenderResponse, error) {
|
||||||
if s.aiRecommendationClient == nil {
|
if s.aiRecommendationClient == nil {
|
||||||
return nil, ErrAIRecommendationNotConfigured
|
return nil, ErrAIRecommendationNotConfigured
|
||||||
}
|
}
|
||||||
|
|
||||||
companyID = strings.TrimSpace(companyID)
|
companyID = strings.TrimSpace(companyID)
|
||||||
if companyID == "" {
|
if companyID == "" {
|
||||||
return nil, errors.New("company ID is required")
|
return nil, errors.New("company ID is required")
|
||||||
}
|
}
|
||||||
|
|
||||||
if cached, ok := s.getCachedAIRecommendations(ctx, companyID); ok {
|
|
||||||
return cached, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
if _, err := s.repository.GetByID(ctx, companyID); err != nil {
|
if _, err := s.repository.GetByID(ctx, companyID); err != nil {
|
||||||
s.logger.Error("Failed to load company for AI recommendations", map[string]interface{}{
|
s.logger.Error("Failed to load company for AI recommendations", map[string]interface{}{
|
||||||
"company_id": companyID,
|
"company_id": companyID,
|
||||||
@@ -144,7 +166,15 @@ func (s *companyService) GetAIRecommendations(ctx context.Context, companyID str
|
|||||||
return nil, errors.New("company not found")
|
return nil, errors.New("company not found")
|
||||||
}
|
}
|
||||||
|
|
||||||
return s.fetchAndCacheAIRecommendations(ctx, companyID)
|
if cached, ok := s.getCachedAIRecommendations(ctx, companyID); ok {
|
||||||
|
return cached, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
s.logger.Debug("AI recommendations cache miss; returning empty list", map[string]interface{}{
|
||||||
|
"company_id": companyID,
|
||||||
|
})
|
||||||
|
|
||||||
|
return []RecommendedTenderResponse{}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *companyService) refreshAIRecommendationsCacheAsync(companyID string) {
|
func (s *companyService) refreshAIRecommendationsCacheAsync(companyID string) {
|
||||||
@@ -186,7 +216,7 @@ func (s *companyService) refreshAIRecommendationsCacheAsync(companyID string) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (s *companyService) fetchAndCacheAIRecommendations(ctx context.Context, companyID string) ([]RecommendedTenderResponse, error) {
|
func (s *companyService) fetchAndCacheAIRecommendations(ctx context.Context, companyID string) ([]RecommendedTenderResponse, error) {
|
||||||
s.logger.Info("Fetching AI tender recommendations", map[string]interface{}{
|
s.logger.Info("Fetching AI tender recommendations from AI service", map[string]interface{}{
|
||||||
"company_id": companyID,
|
"company_id": companyID,
|
||||||
})
|
})
|
||||||
|
|
||||||
@@ -216,7 +246,7 @@ func (s *companyService) fetchAndCacheAIRecommendations(ctx context.Context, com
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (s *companyService) getCachedAIRecommendations(ctx context.Context, companyID string) ([]RecommendedTenderResponse, bool) {
|
func (s *companyService) getCachedAIRecommendations(ctx context.Context, companyID string) ([]RecommendedTenderResponse, bool) {
|
||||||
if s.redisClient == nil || s.recommendationCacheTTL <= 0 {
|
if s.redisClient == nil {
|
||||||
return nil, false
|
return nil, false
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -250,7 +280,7 @@ func (s *companyService) getCachedAIRecommendations(ctx context.Context, company
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (s *companyService) cacheAIRecommendations(ctx context.Context, companyID string, responses []RecommendedTenderResponse) {
|
func (s *companyService) cacheAIRecommendations(ctx context.Context, companyID string, responses []RecommendedTenderResponse) {
|
||||||
if s.redisClient == nil || s.recommendationCacheTTL <= 0 {
|
if s.redisClient == nil {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -263,7 +293,8 @@ func (s *companyService) cacheAIRecommendations(ctx context.Context, companyID s
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
if err := s.redisClient.Set(ctx, aiRecommendationCacheKey(companyID), string(encoded), s.recommendationCacheTTL); err != nil {
|
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{}{
|
s.logger.Warn("Failed to store AI recommendation cache", map[string]interface{}{
|
||||||
"company_id": companyID,
|
"company_id": companyID,
|
||||||
"error": err.Error(),
|
"error": err.Error(),
|
||||||
@@ -271,6 +302,26 @@ func (s *companyService) cacheAIRecommendations(ctx context.Context, companyID s
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
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) {
|
func (s *companyService) buildOnboardingDocuments(fileIDs []string) (string, error) {
|
||||||
if len(fileIDs) == 0 {
|
if len(fileIDs) == 0 {
|
||||||
return "[]", nil
|
return "[]", nil
|
||||||
|
|||||||
Reference in New Issue
Block a user