Enhance tender recommendation service with caching and parallel processing
continuous-integration/drone/push Build is passing
continuous-integration/drone/push Build is passing
- Introduced a caching mechanism for translation responses in the tender recommendation service, improving efficiency by reducing redundant translation requests. - Refactored the `enrichWithTranslation` method to support an optional translation cache, allowing for faster retrieval of stored translations. - Implemented parallel processing for building recommended tender responses, utilizing goroutines to enhance performance and responsiveness. - Added unit tests to validate the new caching behavior and ensure the correct application of translations in various scenarios. This update significantly improves the performance and responsiveness of the tender recommendation service by integrating caching and parallel processing, enhancing the overall user experience.
This commit is contained in:
@@ -191,6 +191,10 @@ func (s *companyService) GetAIRecommendations(ctx context.Context, companyID str
|
||||
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 {
|
||||
s.logger.Error("Failed to load company for AI recommendations", map[string]interface{}{
|
||||
"company_id": companyID,
|
||||
@@ -199,11 +203,7 @@ func (s *companyService) GetAIRecommendations(ctx context.Context, companyID str
|
||||
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{}{
|
||||
s.logger.Debug("AI recommendations cache miss; returning empty list", map[string]interface{}{
|
||||
"company_id": companyID,
|
||||
})
|
||||
|
||||
@@ -234,7 +234,7 @@ func (s *companyService) getCachedAIRecommendationsOrEmpty(ctx context.Context,
|
||||
return cached
|
||||
}
|
||||
|
||||
s.logger.Info("AI recommendations cache miss; returning empty list", map[string]interface{}{
|
||||
s.logger.Debug("AI recommendations cache miss; returning empty list", map[string]interface{}{
|
||||
"company_id": companyID,
|
||||
})
|
||||
return []RecommendedTenderResponse{}
|
||||
@@ -251,12 +251,11 @@ func (s *companyService) GetMergedAIRecommendations(ctx context.Context, company
|
||||
return []RecommendedTenderResponse{}, nil
|
||||
}
|
||||
|
||||
if err := s.validateRecommendationCompanyIDs(ctx, normalized); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
recommendationLists := make([][]RecommendedTenderResponse, len(normalized))
|
||||
group, groupCtx := errgroup.WithContext(ctx)
|
||||
group.Go(func() error {
|
||||
return s.validateRecommendationCompanyIDs(groupCtx, normalized)
|
||||
})
|
||||
for i, companyID := range normalized {
|
||||
i := i
|
||||
companyID := companyID
|
||||
|
||||
@@ -0,0 +1,129 @@
|
||||
package tender
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"golang.org/x/sync/errgroup"
|
||||
|
||||
"tm/pkg/ai_summarizer"
|
||||
)
|
||||
|
||||
const recommendationTranslationConcurrency = 8
|
||||
|
||||
type rankedRecommendedTender struct {
|
||||
tender Tender
|
||||
rank int
|
||||
analysis string
|
||||
tenderRef string
|
||||
}
|
||||
|
||||
type recommendationTranslationCache struct {
|
||||
mu sync.Mutex
|
||||
items map[string]ai_summarizer.StoredTranslation
|
||||
}
|
||||
|
||||
func newRecommendationTranslationCache() *recommendationTranslationCache {
|
||||
return &recommendationTranslationCache{
|
||||
items: make(map[string]ai_summarizer.StoredTranslation),
|
||||
}
|
||||
}
|
||||
|
||||
func recommendationTranslationCacheKey(tenderID, language string) string {
|
||||
return strings.TrimSpace(tenderID) + "|" + strings.TrimSpace(language)
|
||||
}
|
||||
|
||||
func (c *recommendationTranslationCache) get(tenderID, language string) (ai_summarizer.StoredTranslation, bool) {
|
||||
if c == nil {
|
||||
return ai_summarizer.StoredTranslation{}, false
|
||||
}
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
stored, ok := c.items[recommendationTranslationCacheKey(tenderID, language)]
|
||||
return stored, ok
|
||||
}
|
||||
|
||||
func (c *recommendationTranslationCache) set(tenderID, language string, stored ai_summarizer.StoredTranslation) {
|
||||
if c == nil {
|
||||
return
|
||||
}
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
c.items[recommendationTranslationCacheKey(tenderID, language)] = stored
|
||||
}
|
||||
|
||||
func (s *tenderService) buildRecommendedTenderResponsesParallel(
|
||||
ctx context.Context,
|
||||
items []rankedRecommendedTender,
|
||||
language string,
|
||||
) []TenderResponse {
|
||||
if len(items) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
out := make([]TenderResponse, len(items))
|
||||
if len(items) == 1 {
|
||||
out[0] = s.buildRecommendedTenderResponse(ctx, items[0].tender, items[0].rank, items[0].analysis, items[0].tenderRef, language, newRecommendationTranslationCache())
|
||||
return out
|
||||
}
|
||||
|
||||
translationCache := newRecommendationTranslationCache()
|
||||
group, groupCtx := errgroup.WithContext(ctx)
|
||||
group.SetLimit(recommendationTranslationConcurrency)
|
||||
|
||||
for i := range items {
|
||||
i := i
|
||||
item := items[i]
|
||||
group.Go(func() error {
|
||||
out[i] = s.buildRecommendedTenderResponse(
|
||||
groupCtx,
|
||||
item.tender,
|
||||
item.rank,
|
||||
item.analysis,
|
||||
item.tenderRef,
|
||||
language,
|
||||
translationCache,
|
||||
)
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
if err := group.Wait(); err != nil {
|
||||
s.logger.Warn("Parallel recommendation response build degraded to sequential", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
"count": len(items),
|
||||
})
|
||||
for i, item := range items {
|
||||
out[i] = s.buildRecommendedTenderResponse(ctx, item.tender, item.rank, item.analysis, item.tenderRef, language, translationCache)
|
||||
}
|
||||
}
|
||||
|
||||
return out
|
||||
}
|
||||
|
||||
func (s *tenderService) loadRecommendationExclusions(
|
||||
ctx context.Context,
|
||||
form *SearchForm,
|
||||
companyID string,
|
||||
companyIDs []string,
|
||||
) (map[string]struct{}, map[string]map[string]struct{}, error) {
|
||||
if !form.ExcludeRejectedTenders {
|
||||
return nil, nil, nil
|
||||
}
|
||||
|
||||
if len(companyIDs) > 0 {
|
||||
excludedByCompany, err := s.buildRejectedTenderIDSetsForCompanies(ctx, companyIDs)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("failed to build recommendation exclusions: %w", err)
|
||||
}
|
||||
return nil, excludedByCompany, nil
|
||||
}
|
||||
|
||||
excluded, err := s.buildRejectedTenderIDSet(ctx, companyID)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("failed to build recommendation exclusions: %w", err)
|
||||
}
|
||||
return excluded, nil, nil
|
||||
}
|
||||
@@ -0,0 +1,90 @@
|
||||
package tender
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"go.mongodb.org/mongo-driver/v2/bson"
|
||||
|
||||
"tm/pkg/ai_summarizer"
|
||||
)
|
||||
|
||||
type countingTranslationStorage struct {
|
||||
calls atomic.Int32
|
||||
}
|
||||
|
||||
func (s *countingTranslationStorage) GetSummaryFromStorage(context.Context, string, string) (string, error) {
|
||||
return "", ai_summarizer.ErrSummaryNotReady
|
||||
}
|
||||
|
||||
func (s *countingTranslationStorage) GetDocumentSummariesFromStorage(context.Context, string, string) ([]ai_summarizer.DocumentSummary, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (s *countingTranslationStorage) GetTranslationFromStorage(context.Context, string, string, string) (ai_summarizer.StoredTranslation, error) {
|
||||
s.calls.Add(1)
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
return ai_summarizer.StoredTranslation{
|
||||
Title: "Translated title",
|
||||
Description: "Translated description",
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *countingTranslationStorage) ListDocuments(context.Context, string, string) ([]ai_summarizer.StoredDocument, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (s *countingTranslationStorage) DownloadDocument(context.Context, string) (io.ReadCloser, error) {
|
||||
return nil, ai_summarizer.ErrObjectNotFound
|
||||
}
|
||||
|
||||
func TestBuildRecommendedTenderResponsesParallelPreservesOrder(t *testing.T) {
|
||||
storage := &countingTranslationStorage{}
|
||||
service := &tenderService{
|
||||
aiSummarizerStorage: storage,
|
||||
defaultLanguage: "en",
|
||||
logger: noopTenderTestLogger{},
|
||||
}
|
||||
|
||||
items := make([]rankedRecommendedTender, 0, 4)
|
||||
ids := []string{
|
||||
"507f1f77bcf86cd799439011",
|
||||
"507f1f77bcf86cd799439012",
|
||||
"507f1f77bcf86cd799439013",
|
||||
"507f1f77bcf86cd799439014",
|
||||
}
|
||||
for i, idHex := range ids {
|
||||
tender := Tender{
|
||||
ContractFolderID: "folder",
|
||||
NoticePublicationID: "notice",
|
||||
Title: "Original",
|
||||
}
|
||||
tender.ID, _ = bson.ObjectIDFromHex(idHex)
|
||||
items = append(items, rankedRecommendedTender{
|
||||
tender: tender,
|
||||
rank: i + 1,
|
||||
analysis: "fit",
|
||||
tenderRef: "PROC_folder/notice",
|
||||
})
|
||||
}
|
||||
|
||||
got := service.buildRecommendedTenderResponsesParallel(context.Background(), items, "fr")
|
||||
if len(got) != len(items) {
|
||||
t.Fatalf("len = %d, want %d", len(got), len(items))
|
||||
}
|
||||
for i, resp := range got {
|
||||
wantRank := i + 1
|
||||
if resp.Rank != wantRank {
|
||||
t.Fatalf("got[%d].Rank = %d, want %d", i, resp.Rank, wantRank)
|
||||
}
|
||||
if resp.Title != "Translated title" {
|
||||
t.Fatalf("got[%d].Title = %q, want translated title", i, resp.Title)
|
||||
}
|
||||
}
|
||||
if storage.calls.Load() != int32(len(items)) {
|
||||
t.Fatalf("translation lookups = %d, want %d", storage.calls.Load(), len(items))
|
||||
}
|
||||
}
|
||||
@@ -77,6 +77,7 @@ func TestBuildRecommendedTenderResponseAppliesStoredTranslation(t *testing.T) {
|
||||
"Strong fit",
|
||||
"PROC_folder-1/00123456-2026",
|
||||
"fr",
|
||||
nil,
|
||||
)
|
||||
|
||||
if resp.Title != "Titre traduit" {
|
||||
@@ -120,6 +121,7 @@ func TestBuildRecommendedTenderResponseKeepsOriginalTextWithoutStoredTranslation
|
||||
"Fallback",
|
||||
"PROC_folder-1/00123456-2026",
|
||||
"fr",
|
||||
nil,
|
||||
)
|
||||
|
||||
if resp.Title != "Original title" {
|
||||
|
||||
+36
-28
@@ -176,7 +176,7 @@ func (s *tenderService) GetByIDs(ctx context.Context, ids []string) (map[string]
|
||||
// buildDetailResponse maps a tender to API form and enriches translation (MinIO) and summary.
|
||||
func (s *tenderService) buildDetailResponse(ctx context.Context, tender *Tender, language string) *TenderResponse {
|
||||
resp := tender.ToResponseWithLanguage("")
|
||||
s.enrichWithTranslation(ctx, tender, resp, language)
|
||||
s.enrichWithTranslation(ctx, tender, resp, language, nil)
|
||||
s.enrichWithAISummary(ctx, tender, resp)
|
||||
return resp
|
||||
}
|
||||
@@ -206,7 +206,7 @@ func (s *tenderService) enrichWithAISummary(ctx context.Context, t *Tender, resp
|
||||
|
||||
// enrichWithTranslation overlays title/description from MinIO when the pipeline has finished
|
||||
// for the requested language. Does not call the AI translate API on GET.
|
||||
func (s *tenderService) enrichWithTranslation(ctx context.Context, t *Tender, resp *TenderResponse, language string) {
|
||||
func (s *tenderService) enrichWithTranslation(ctx context.Context, t *Tender, resp *TenderResponse, language string, cache *recommendationTranslationCache) {
|
||||
if t == nil || resp == nil {
|
||||
return
|
||||
}
|
||||
@@ -215,6 +215,17 @@ func (s *tenderService) enrichWithTranslation(ctx context.Context, t *Tender, re
|
||||
}
|
||||
|
||||
targetLanguage := s.resolveDetailLanguage(language)
|
||||
if cache != nil {
|
||||
if stored, ok := cache.get(t.GetID(), targetLanguage); ok {
|
||||
if strings.TrimSpace(stored.Title) != "" {
|
||||
resp.Title = stored.Title
|
||||
resp.Description = stored.Description
|
||||
resp.Language = targetLanguage
|
||||
}
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
enrichCtx, cancel := context.WithTimeout(ctx, aiTranslationEnrichmentTimeout)
|
||||
defer cancel()
|
||||
|
||||
@@ -233,6 +244,10 @@ func (s *tenderService) enrichWithTranslation(ctx context.Context, t *Tender, re
|
||||
return
|
||||
}
|
||||
|
||||
if cache != nil {
|
||||
cache.set(t.GetID(), targetLanguage, stored)
|
||||
}
|
||||
|
||||
resp.Title = stored.Title
|
||||
resp.Description = stored.Description
|
||||
resp.Language = targetLanguage
|
||||
@@ -947,7 +962,7 @@ func (s *tenderService) Recommend(ctx context.Context, form *SearchForm, paginat
|
||||
return nil, fmt.Errorf("%w: cursor pagination is not supported for recommendations", response.ErrInvalidPagination)
|
||||
}
|
||||
|
||||
s.logger.Info("Recommend tenders via AI service", map[string]interface{}{
|
||||
s.logger.Debug("Recommend tenders request", map[string]interface{}{
|
||||
"company_id": companyID,
|
||||
"company_ids": companyIDs,
|
||||
})
|
||||
@@ -968,39 +983,35 @@ func (s *tenderService) Recommend(ctx context.Context, form *SearchForm, paginat
|
||||
return s.emptySearchResponse(pagination), nil
|
||||
}
|
||||
|
||||
tenderByRef, err := s.resolveRecommendedTenders(ctx, recommendations)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var (
|
||||
tenderByRef map[string]Tender
|
||||
excluded map[string]struct{}
|
||||
excludedByCompany map[string]map[string]struct{}
|
||||
)
|
||||
if form.ExcludeRejectedTenders {
|
||||
if len(companyIDs) > 0 {
|
||||
excludedByCompany, err = s.buildRejectedTenderIDSetsForCompanies(ctx, companyIDs)
|
||||
} else {
|
||||
excluded, err = s.buildRejectedTenderIDSet(ctx, companyID)
|
||||
}
|
||||
if err != nil {
|
||||
loadGroup, loadCtx := errgroup.WithContext(ctx)
|
||||
loadGroup.Go(func() error {
|
||||
var resolveErr error
|
||||
tenderByRef, resolveErr = s.resolveRecommendedTenders(loadCtx, recommendations)
|
||||
return resolveErr
|
||||
})
|
||||
loadGroup.Go(func() error {
|
||||
var exclusionErr error
|
||||
excluded, excludedByCompany, exclusionErr = s.loadRecommendationExclusions(loadCtx, form, companyID, companyIDs)
|
||||
return exclusionErr
|
||||
})
|
||||
if err := loadGroup.Wait(); err != nil {
|
||||
if form.ExcludeRejectedTenders {
|
||||
s.logger.Error("Failed to build recommendation exclusions", map[string]interface{}{
|
||||
"company_id": companyID,
|
||||
"company_ids": companyIDs,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil, fmt.Errorf("failed to build recommendation exclusions: %w", err)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
|
||||
lang := s.pickResponseLanguage(form.Language)
|
||||
now := time.Now().Unix()
|
||||
type rankedRecommendedTender struct {
|
||||
tender Tender
|
||||
rank int
|
||||
analysis string
|
||||
tenderRef string
|
||||
}
|
||||
ordered := make([]rankedRecommendedTender, 0, len(recommendations))
|
||||
seenTenderIDs := make(map[string]struct{}, len(recommendations))
|
||||
for _, rec := range recommendations {
|
||||
@@ -1044,10 +1055,7 @@ func (s *tenderService) Recommend(ctx context.Context, form *SearchForm, paginat
|
||||
end = len(ordered)
|
||||
}
|
||||
|
||||
paged := make([]TenderResponse, 0, end-start)
|
||||
for _, item := range ordered[start:end] {
|
||||
paged = append(paged, s.buildRecommendedTenderResponse(ctx, item.tender, item.rank, item.analysis, item.tenderRef, lang))
|
||||
}
|
||||
paged := s.buildRecommendedTenderResponsesParallel(ctx, ordered[start:end], lang)
|
||||
|
||||
return &SearchResponse{
|
||||
Tenders: paged,
|
||||
@@ -1055,9 +1063,9 @@ func (s *tenderService) Recommend(ctx context.Context, form *SearchForm, paginat
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *tenderService) buildRecommendedTenderResponse(ctx context.Context, tender Tender, rank int, analysis, tenderRef, language string) TenderResponse {
|
||||
func (s *tenderService) buildRecommendedTenderResponse(ctx context.Context, tender Tender, rank int, analysis, tenderRef, language string, cache *recommendationTranslationCache) TenderResponse {
|
||||
resp := tender.ToResponseWithLanguage(language)
|
||||
s.enrichWithTranslation(ctx, &tender, resp, language)
|
||||
s.enrichWithTranslation(ctx, &tender, resp, language, cache)
|
||||
resp.Rank = rank
|
||||
resp.Analysis = analysis
|
||||
resp.ProcedureRef = tenderRef
|
||||
|
||||
Reference in New Issue
Block a user