Compare commits
2 Commits
f108733c2a
...
0b74e9ad23
| Author | SHA1 | Date | |
|---|---|---|---|
| 0b74e9ad23 | |||
| b28bc23975 |
@@ -73,6 +73,8 @@ type AISummarizerConfig struct {
|
||||
// RecommendationCacheTTL is an optional max Redis TTL for cached recommendations.
|
||||
// When 0 (default), entries persist until the company is created, updated, or onboarded.
|
||||
RecommendationCacheTTL time.Duration `env:"AI_RECOMMENDATION_CACHE_TTL" envDefault:"0"`
|
||||
// RecommendationPageCacheLanguages lists languages pre-built for the recommendation page cache.
|
||||
RecommendationPageCacheLanguages string `env:"AI_RECOMMENDATION_PAGE_CACHE_LANGUAGES" envDefault:"en"`
|
||||
|
||||
// MinIO storage settings
|
||||
MinioEndpoint string `env:"AI_SUMMARIZER_MINIO_ENDPOINT" envDefault:""`
|
||||
|
||||
+26
-1
@@ -241,8 +241,33 @@ func main() {
|
||||
userService := user.NewService(userRepository, logger, userAuthService, userValidator, notificationSDK, redisClient, auditLogger)
|
||||
categoryService := company_category.NewService(categoryRepository, logger)
|
||||
companyService := company.NewService(companyRepository, categoryService, fileStoreService, aiRecommendationClient, aiPipelineClient, redisClient, conf.AISummarizer.RecommendationCacheTTL, logger)
|
||||
pageCacheLanguages := tender.ParsePageCacheLanguagesConfig(conf.AISummarizer.RecommendationPageCacheLanguages)
|
||||
if configurer, ok := companyService.(interface {
|
||||
SetPageCacheLanguages([]string)
|
||||
}); ok {
|
||||
configurer.SetPageCacheLanguages(pageCacheLanguages)
|
||||
}
|
||||
customerService := customer.NewService(customerRepository, logger, customerAuthService, companyService, redisClient, notificationSDK, customerValidator, auditLogger)
|
||||
tenderService := tender.NewService(tenderRepository, companyService, customerService, tenderApprovalRepository, logger, aiSummarizerClient, aiSummarizerStorage, conf.AISummarizer.DefaultLanguage)
|
||||
tenderService := tender.NewService(
|
||||
tenderRepository,
|
||||
companyService,
|
||||
customerService,
|
||||
tenderApprovalRepository,
|
||||
logger,
|
||||
aiSummarizerClient,
|
||||
aiSummarizerStorage,
|
||||
redisClient,
|
||||
conf.AISummarizer.RecommendationCacheTTL,
|
||||
pageCacheLanguages,
|
||||
conf.AISummarizer.DefaultLanguage,
|
||||
)
|
||||
if configurer, ok := companyService.(interface {
|
||||
SetRecommendedTendersPageCacheRefresher(company.RecommendedTendersPageCacheRefresher)
|
||||
}); ok {
|
||||
if refresher, ok := tenderService.(company.RecommendedTendersPageCacheRefresher); ok {
|
||||
configurer.SetRecommendedTendersPageCacheRefresher(refresher)
|
||||
}
|
||||
}
|
||||
feedbackService := feedback.NewService(feedbackRepo, tenderService, companyService, logger)
|
||||
tenderApprovalService := tender_approval.NewService(tenderApprovalRepository, tenderService, logger)
|
||||
inquiryService := inquiry.NewService(inquiryRepository, notificationSDK, logger, xssPolicy)
|
||||
|
||||
@@ -121,6 +121,11 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger
|
||||
config.AISummarizer.RecommendationCacheTTL,
|
||||
appLogger,
|
||||
)
|
||||
if configurer, ok := recommendationRefresher.(interface {
|
||||
SetPageCacheLanguages([]string)
|
||||
}); ok {
|
||||
configurer.SetPageCacheLanguages(tender.ParsePageCacheLanguagesConfig(config.AISummarizer.RecommendationPageCacheLanguages))
|
||||
}
|
||||
appLogger.Info("Company recommendation refresh after pipeline enabled", map[string]interface{}{})
|
||||
} else if config.Worker.RecommendationRefreshAfterPipelineEnabled {
|
||||
appLogger.Warn("Company recommendation refresh after pipeline disabled: AI client or Redis unavailable", map[string]interface{}{})
|
||||
|
||||
@@ -60,4 +60,5 @@ type AISummarizerConfig struct {
|
||||
MinioRegion string `env:"AI_SUMMARIZER_MINIO_REGION" envDefault:"us-east-1"`
|
||||
MinioBucket string `env:"AI_SUMMARIZER_MINIO_BUCKET" envDefault:"opplens"`
|
||||
RecommendationCacheTTL time.Duration `env:"AI_RECOMMENDATION_CACHE_TTL" envDefault:"0"`
|
||||
RecommendationPageCacheLanguages string `env:"AI_RECOMMENDATION_PAGE_CACHE_LANGUAGES" envDefault:"en"`
|
||||
}
|
||||
|
||||
@@ -0,0 +1,7 @@
|
||||
package company
|
||||
|
||||
// RecommendedTendersPageCacheRefresher pre-builds resolved recommendation pages in Redis.
|
||||
// Implemented by the tender service on the web API; optional on the worker.
|
||||
type RecommendedTendersPageCacheRefresher interface {
|
||||
ScheduleRefreshRecommendedTendersPageCache(companyID string)
|
||||
}
|
||||
@@ -464,7 +464,11 @@ func (s *companyService) cacheAIRecommendations(ctx context.Context, companyID s
|
||||
"company_id": companyID,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
s.invalidateRecommendedTendersPageCache(ctx, companyID)
|
||||
s.scheduleRecommendedTendersPageCacheRefresh(companyID)
|
||||
}
|
||||
|
||||
func (s *companyService) recommendationCacheExpiration() time.Duration {
|
||||
@@ -485,6 +489,7 @@ func (s *companyService) invalidateAIRecommendationCache(ctx context.Context, co
|
||||
"error": err.Error(),
|
||||
})
|
||||
}
|
||||
s.invalidateRecommendedTendersPageCache(ctx, companyID)
|
||||
}
|
||||
|
||||
func (s *companyService) buildOnboardingDocuments(fileIDs []string) (string, error) {
|
||||
|
||||
@@ -0,0 +1,71 @@
|
||||
package company
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
)
|
||||
|
||||
const recommendedTendersPageCacheKeyPrefix = "ai:recommendations:page:"
|
||||
|
||||
func recommendedTendersPageCacheKey(companyID, language string) string {
|
||||
return recommendedTendersPageCacheKeyPrefix +
|
||||
strings.TrimSpace(companyID) + ":" +
|
||||
strings.ToLower(strings.TrimSpace(language))
|
||||
}
|
||||
|
||||
func (s *companyService) invalidateRecommendedTendersPageCache(ctx context.Context, companyID string) {
|
||||
if s.redisClient == nil {
|
||||
return
|
||||
}
|
||||
|
||||
for _, language := range s.pageCacheLanguages {
|
||||
lang := strings.ToLower(strings.TrimSpace(language))
|
||||
if lang == "" {
|
||||
continue
|
||||
}
|
||||
_ = s.redisClient.Del(ctx, recommendedTendersPageCacheKey(companyID, lang))
|
||||
}
|
||||
}
|
||||
|
||||
func (s *companyService) scheduleRecommendedTendersPageCacheRefresh(companyID string) {
|
||||
if s.pageCacheRefresher == nil {
|
||||
return
|
||||
}
|
||||
s.pageCacheRefresher.ScheduleRefreshRecommendedTendersPageCache(companyID)
|
||||
}
|
||||
|
||||
// SetRecommendedTendersPageCacheRefresher wires async page-cache rebuilds from the tender service.
|
||||
func (s *companyService) SetRecommendedTendersPageCacheRefresher(refresher RecommendedTendersPageCacheRefresher) {
|
||||
s.pageCacheRefresher = refresher
|
||||
}
|
||||
|
||||
// SetPageCacheLanguages configures which language variants are invalidated in Redis.
|
||||
func (s *companyService) SetPageCacheLanguages(languages []string) {
|
||||
s.pageCacheLanguages = normalizePageCacheLanguages("", languages)
|
||||
}
|
||||
|
||||
func normalizePageCacheLanguages(defaultLanguage string, languages []string) []string {
|
||||
seen := make(map[string]struct{})
|
||||
out := make([]string, 0, len(languages)+1)
|
||||
|
||||
add := func(lang string) {
|
||||
lang = strings.ToLower(strings.TrimSpace(lang))
|
||||
if lang == "" {
|
||||
return
|
||||
}
|
||||
if _, ok := seen[lang]; ok {
|
||||
return
|
||||
}
|
||||
seen[lang] = struct{}{}
|
||||
out = append(out, lang)
|
||||
}
|
||||
|
||||
add(defaultLanguage)
|
||||
for _, lang := range languages {
|
||||
add(lang)
|
||||
}
|
||||
if len(out) == 0 {
|
||||
out = append(out, "en")
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -60,8 +60,10 @@ type companyService struct {
|
||||
fileStore filestore.FileStoreService
|
||||
aiRecommendationClient AIRecommendationClient
|
||||
aiPipelineStatusClient AIPipelineStatusClient
|
||||
pageCacheRefresher RecommendedTendersPageCacheRefresher
|
||||
redisClient redis.Client
|
||||
recommendationCacheTTL time.Duration
|
||||
pageCacheLanguages []string
|
||||
logger logger.Logger
|
||||
}
|
||||
|
||||
|
||||
@@ -1,6 +1,10 @@
|
||||
package tender
|
||||
|
||||
import "testing"
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"go.mongodb.org/mongo-driver/v2/bson"
|
||||
)
|
||||
|
||||
func TestParseAIProcedureRef(t *testing.T) {
|
||||
tests := []struct {
|
||||
@@ -84,3 +88,37 @@ func TestMapProcedureReferencesToTenders(t *testing.T) {
|
||||
t.Fatalf("expected folder-2 ref to resolve, got %+v", got["PROC_folder-2/notice-2"])
|
||||
}
|
||||
}
|
||||
|
||||
func TestMapProcedureReferencesToTendersUsesFolderCandidatesWhenNoticeLookupMisses(t *testing.T) {
|
||||
// Notice-old is only present on the merged tender loaded via contract_folder_id.
|
||||
candidates := []Tender{
|
||||
{
|
||||
ContractFolderID: "folder-1",
|
||||
NoticePublicationID: "notice-new",
|
||||
RelatedNoticePublicationIDs: []string{"notice-old"},
|
||||
},
|
||||
}
|
||||
refs := []ProcedureReference{
|
||||
{Ref: "PROC_folder-1/notice-old", ContractFolderID: "folder-1", NoticePublicationID: "notice-old"},
|
||||
}
|
||||
|
||||
got := mapProcedureReferencesToTenders(candidates, refs)
|
||||
if len(got) != 1 {
|
||||
t.Fatalf("mapProcedureReferencesToTenders() len = %d, want 1", len(got))
|
||||
}
|
||||
if got["PROC_folder-1/notice-old"].NoticePublicationID != "notice-new" {
|
||||
t.Fatalf("expected folder candidate to resolve notice-old ref, got %+v", got["PROC_folder-1/notice-old"])
|
||||
}
|
||||
}
|
||||
|
||||
func TestMergeTenderCandidatesDedupesByMongoID(t *testing.T) {
|
||||
first := Tender{ContractFolderID: "folder-1", NoticePublicationID: "notice-a"}
|
||||
first.ID, _ = bson.ObjectIDFromHex("507f1f77bcf86cd799439011")
|
||||
second := Tender{ContractFolderID: "folder-1", NoticePublicationID: "notice-b"}
|
||||
second.ID = first.ID
|
||||
|
||||
got := mergeTenderCandidates([]Tender{first}, []Tender{second})
|
||||
if len(got) != 1 {
|
||||
t.Fatalf("mergeTenderCandidates() len = %d, want 1", len(got))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -97,19 +97,22 @@ func addTenderIDsToSet(set map[string]struct{}, ids []string) {
|
||||
// markResolvedRecommendedTenderSeen records a resolved tender and reports whether it was already seen.
|
||||
// Recommendations may reference the same tender via different AI refs (procedure ref vs Mongo ID,
|
||||
// or multiple notice publication IDs for one merged tender).
|
||||
func markResolvedRecommendedTenderSeen(seen map[string]struct{}, tender Tender) bool {
|
||||
func markResolvedRecommendedTenderSeen(seen map[string]struct{}, tender Tender, tenderRef string) bool {
|
||||
if seen == nil {
|
||||
return false
|
||||
}
|
||||
|
||||
tenderID := strings.TrimSpace(tender.GetID())
|
||||
if tenderID == "" {
|
||||
key := strings.TrimSpace(tender.GetID())
|
||||
if key == "" {
|
||||
key = strings.TrimSpace(tenderRef)
|
||||
}
|
||||
if key == "" {
|
||||
return false
|
||||
}
|
||||
if _, exists := seen[key]; exists {
|
||||
return true
|
||||
}
|
||||
if _, exists := seen[tenderID]; exists {
|
||||
return true
|
||||
}
|
||||
seen[tenderID] = struct{}{}
|
||||
seen[key] = struct{}{}
|
||||
return false
|
||||
}
|
||||
|
||||
|
||||
@@ -132,8 +132,8 @@ func TestMarkResolvedRecommendedTenderSeen(t *testing.T) {
|
||||
sample.ID, _ = bson.ObjectIDFromHex("507f1f77bcf86cd799439011")
|
||||
|
||||
seen := make(map[string]struct{})
|
||||
first := markResolvedRecommendedTenderSeen(seen, sample)
|
||||
second := markResolvedRecommendedTenderSeen(seen, sample)
|
||||
first := markResolvedRecommendedTenderSeen(seen, sample, "PROC_folder/notice")
|
||||
second := markResolvedRecommendedTenderSeen(seen, sample, "PROC_folder/notice")
|
||||
if first {
|
||||
t.Fatalf("first sighting should not be skipped, id=%q seen=%v", sample.GetID(), seen)
|
||||
}
|
||||
@@ -143,7 +143,7 @@ func TestMarkResolvedRecommendedTenderSeen(t *testing.T) {
|
||||
|
||||
otherRefSameTender := Tender{}
|
||||
otherRefSameTender.ID = sample.ID
|
||||
if !markResolvedRecommendedTenderSeen(seen, otherRefSameTender) {
|
||||
if !markResolvedRecommendedTenderSeen(seen, otherRefSameTender, "PROC_folder/other-notice") {
|
||||
t.Fatalf("duplicate tender via different ref should be skipped")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -69,7 +69,7 @@ func (s *tenderService) buildRecommendationPage(
|
||||
if len(form.Status) > 0 && !statusAllowed(tender.Status, form.Status) {
|
||||
continue
|
||||
}
|
||||
if markResolvedRecommendedTenderSeen(seenTenderIDs, tender) {
|
||||
if markResolvedRecommendedTenderSeen(seenTenderIDs, tender, tenderRef) {
|
||||
continue
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,423 @@
|
||||
package tender
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
"go.mongodb.org/mongo-driver/v2/bson"
|
||||
"golang.org/x/sync/errgroup"
|
||||
|
||||
"tm/internal/company"
|
||||
"tm/pkg/response"
|
||||
)
|
||||
|
||||
const (
|
||||
recommendedTendersPageCacheKeyPrefix = "ai:recommendations:page:"
|
||||
recommendationPageCacheReadTimeout = 3 * time.Second
|
||||
recommendationPageCacheBuildTimeout = 5 * time.Minute
|
||||
recommendationPageCacheConcurrency = 8
|
||||
)
|
||||
|
||||
type cachedRecommendedTendersPage struct {
|
||||
BuiltAt int64 `json:"built_at"`
|
||||
CompanyID string `json:"company_id"`
|
||||
Language string `json:"language"`
|
||||
Items []TenderResponse `json:"items"`
|
||||
}
|
||||
|
||||
func recommendedTendersPageCacheKey(companyID, language string) string {
|
||||
return recommendedTendersPageCacheKeyPrefix +
|
||||
strings.TrimSpace(companyID) + ":" +
|
||||
strings.ToLower(strings.TrimSpace(language))
|
||||
}
|
||||
|
||||
// ScheduleRefreshRecommendedTendersPageCache rebuilds resolved recommendation pages asynchronously.
|
||||
func (s *tenderService) ScheduleRefreshRecommendedTendersPageCache(companyID string) {
|
||||
companyID = strings.TrimSpace(companyID)
|
||||
if companyID == "" || s.redisClient == nil {
|
||||
return
|
||||
}
|
||||
|
||||
go func() {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), recommendationPageCacheBuildTimeout)
|
||||
defer cancel()
|
||||
|
||||
for _, language := range s.pageCacheLanguages {
|
||||
lang := strings.ToLower(strings.TrimSpace(language))
|
||||
if lang == "" {
|
||||
continue
|
||||
}
|
||||
if err := s.buildRecommendedTendersPageCache(ctx, companyID, lang); err != nil {
|
||||
s.logger.Warn("Failed to refresh recommended tenders page cache", map[string]interface{}{
|
||||
"company_id": companyID,
|
||||
"language": lang,
|
||||
"error": err.Error(),
|
||||
})
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
func (s *tenderService) buildRecommendedTendersPageCache(ctx context.Context, companyID, language string) error {
|
||||
recommendations, err := s.companyService.GetAIRecommendations(ctx, companyID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if len(recommendations) == 0 {
|
||||
s.deleteRecommendedTendersPageCache(ctx, companyID, language)
|
||||
return nil
|
||||
}
|
||||
|
||||
items, err := s.buildAllRankedRecommendations(ctx, recommendations)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if len(items) == 0 {
|
||||
s.deleteRecommendedTendersPageCache(ctx, companyID, language)
|
||||
return nil
|
||||
}
|
||||
|
||||
responses := s.buildRecommendedTenderResponsesParallel(ctx, items, language)
|
||||
page := cachedRecommendedTendersPage{
|
||||
BuiltAt: time.Now().Unix(),
|
||||
CompanyID: companyID,
|
||||
Language: language,
|
||||
Items: responses,
|
||||
}
|
||||
|
||||
encoded, err := json.Marshal(page)
|
||||
if err != nil {
|
||||
return fmt.Errorf("encode recommended tenders page cache: %w", err)
|
||||
}
|
||||
|
||||
writeCtx, cancel := context.WithTimeout(ctx, recommendationPageCacheReadTimeout)
|
||||
defer cancel()
|
||||
|
||||
if err := s.redisClient.Set(writeCtx, recommendedTendersPageCacheKey(companyID, language), string(encoded), s.pageCacheTTL); err != nil {
|
||||
return fmt.Errorf("store recommended tenders page cache: %w", err)
|
||||
}
|
||||
|
||||
s.logger.Info("Recommended tenders page cache built", map[string]interface{}{
|
||||
"company_id": companyID,
|
||||
"language": language,
|
||||
"count": len(responses),
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *tenderService) getRecommendedTendersPageCache(ctx context.Context, companyID, language string) ([]TenderResponse, bool) {
|
||||
if s.redisClient == nil {
|
||||
return nil, false
|
||||
}
|
||||
|
||||
readCtx, cancel := context.WithTimeout(ctx, recommendationPageCacheReadTimeout)
|
||||
defer cancel()
|
||||
|
||||
raw, err := s.redisClient.Get(readCtx, recommendedTendersPageCacheKey(companyID, language))
|
||||
if err != nil {
|
||||
if err != redis.Nil {
|
||||
s.logger.Warn("Failed to read recommended tenders page cache", map[string]interface{}{
|
||||
"company_id": companyID,
|
||||
"language": language,
|
||||
"error": err.Error(),
|
||||
})
|
||||
}
|
||||
return nil, false
|
||||
}
|
||||
|
||||
var page cachedRecommendedTendersPage
|
||||
if err := json.Unmarshal([]byte(raw), &page); err != nil {
|
||||
s.logger.Warn("Failed to decode recommended tenders page cache", map[string]interface{}{
|
||||
"company_id": companyID,
|
||||
"language": language,
|
||||
"error": err.Error(),
|
||||
})
|
||||
s.deleteRecommendedTendersPageCache(ctx, companyID, language)
|
||||
return nil, false
|
||||
}
|
||||
if len(page.Items) == 0 {
|
||||
return nil, false
|
||||
}
|
||||
|
||||
return page.Items, true
|
||||
}
|
||||
|
||||
func (s *tenderService) deleteRecommendedTendersPageCache(ctx context.Context, companyID, language string) {
|
||||
if s.redisClient == nil {
|
||||
return
|
||||
}
|
||||
_ = s.redisClient.Del(ctx, recommendedTendersPageCacheKey(companyID, language))
|
||||
}
|
||||
|
||||
func (s *tenderService) buildAllRankedRecommendations(
|
||||
ctx context.Context,
|
||||
recommendations []company.RecommendedTenderResponse,
|
||||
) ([]rankedRecommendedTender, error) {
|
||||
seenTenderIDs := make(map[string]struct{}, len(recommendations))
|
||||
out := make([]rankedRecommendedTender, 0, len(recommendations))
|
||||
|
||||
for start := 0; start < len(recommendations); start += recommendationResolveBatchSize {
|
||||
end := start + recommendationResolveBatchSize
|
||||
if end > len(recommendations) {
|
||||
end = len(recommendations)
|
||||
}
|
||||
|
||||
tenderByRef, err := s.resolveRecommendedTenders(ctx, recommendations[start:end])
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
for _, rec := range recommendations[start:end] {
|
||||
tenderRef := strings.TrimSpace(rec.TenderID)
|
||||
tender, ok := tenderByRef[tenderRef]
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
if markResolvedRecommendedTenderSeen(seenTenderIDs, tender, tenderRef) {
|
||||
continue
|
||||
}
|
||||
|
||||
out = append(out, rankedRecommendedTender{
|
||||
tender: tender,
|
||||
rank: rec.Rank,
|
||||
analysis: rec.Analysis,
|
||||
tenderRef: tenderRef,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
return out, nil
|
||||
}
|
||||
|
||||
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(recommendationPageCacheConcurrency)
|
||||
|
||||
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 page cache 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) recommendFromPageCache(
|
||||
ctx context.Context,
|
||||
companyID string,
|
||||
companyIDs []string,
|
||||
form *SearchForm,
|
||||
pagination *response.Pagination,
|
||||
lang string,
|
||||
) (*SearchResponse, bool) {
|
||||
var cached []TenderResponse
|
||||
if len(companyIDs) > 0 {
|
||||
cached = s.mergeCachedRecommendationPages(ctx, companyIDs, lang)
|
||||
} else {
|
||||
items, ok := s.getRecommendedTendersPageCache(ctx, companyID, lang)
|
||||
if !ok {
|
||||
return nil, false
|
||||
}
|
||||
cached = items
|
||||
}
|
||||
if len(cached) == 0 {
|
||||
return nil, false
|
||||
}
|
||||
|
||||
excluded, excludedByCompany, err := s.loadRecommendationExclusions(ctx, form, companyID, companyIDs)
|
||||
if err != nil {
|
||||
return nil, false
|
||||
}
|
||||
|
||||
now := time.Now().Unix()
|
||||
filtered := filterCachedRecommendedTenders(cached, form, companyIDs, excluded, excludedByCompany, now)
|
||||
total := int64(len(filtered))
|
||||
|
||||
start := pagination.Offset
|
||||
if start > len(filtered) {
|
||||
start = len(filtered)
|
||||
}
|
||||
end := start + pagination.Limit
|
||||
if end > len(filtered) {
|
||||
end = len(filtered)
|
||||
}
|
||||
|
||||
return &SearchResponse{
|
||||
Tenders: append([]TenderResponse(nil), filtered[start:end]...),
|
||||
Metadata: pagination.ListMeta(total, "", end < len(filtered), start),
|
||||
}, true
|
||||
}
|
||||
|
||||
func (s *tenderService) mergeCachedRecommendationPages(ctx context.Context, companyIDs []string, lang string) []TenderResponse {
|
||||
normalized := normalizeRecommendationCompanyScope(companyIDs)
|
||||
if len(normalized) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
lists := make([][]TenderResponse, 0, len(normalized))
|
||||
for _, companyID := range normalized {
|
||||
items, ok := s.getRecommendedTendersPageCache(ctx, companyID, lang)
|
||||
if !ok || len(items) == 0 {
|
||||
return nil
|
||||
}
|
||||
lists = append(lists, items)
|
||||
}
|
||||
|
||||
merged := make([]TenderResponse, 0)
|
||||
for _, list := range lists {
|
||||
merged = append(merged, list...)
|
||||
}
|
||||
return dedupeCachedRecommendedTendersByID(merged)
|
||||
}
|
||||
|
||||
func dedupeCachedRecommendedTendersByID(items []TenderResponse) []TenderResponse {
|
||||
seen := make(map[string]struct{}, len(items))
|
||||
out := make([]TenderResponse, 0, len(items))
|
||||
for _, item := range items {
|
||||
key := strings.TrimSpace(item.ID)
|
||||
if key == "" {
|
||||
key = strings.TrimSpace(item.ProcedureRef)
|
||||
}
|
||||
if key == "" {
|
||||
continue
|
||||
}
|
||||
if _, exists := seen[key]; exists {
|
||||
continue
|
||||
}
|
||||
seen[key] = struct{}{}
|
||||
out = append(out, item)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func filterCachedRecommendedTenders(
|
||||
items []TenderResponse,
|
||||
form *SearchForm,
|
||||
companyIDs []string,
|
||||
excluded map[string]struct{},
|
||||
excludedByCompany map[string]map[string]struct{},
|
||||
now int64,
|
||||
) []TenderResponse {
|
||||
if len(items) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
out := make([]TenderResponse, 0, len(items))
|
||||
for _, item := range items {
|
||||
tender := tenderFromRecommendationResponse(item)
|
||||
tenderRef := strings.TrimSpace(item.ProcedureRef)
|
||||
|
||||
if len(companyIDs) > 0 {
|
||||
if isRecommendedTenderExcludedForAllCompanies(tender, tenderRef, excludedByCompany, companyIDs) {
|
||||
continue
|
||||
}
|
||||
} else if isRecommendedTenderExcluded(tender, tenderRef, excluded) {
|
||||
continue
|
||||
}
|
||||
if form.OnlyActiveDeadlines && !tender.IsRecommendable(now) {
|
||||
continue
|
||||
}
|
||||
if len(form.Status) > 0 && !statusAllowed(tender.Status, form.Status) {
|
||||
continue
|
||||
}
|
||||
|
||||
out = append(out, item)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func tenderFromRecommendationResponse(item TenderResponse) Tender {
|
||||
var tender Tender
|
||||
if id, err := bson.ObjectIDFromHex(strings.TrimSpace(item.ID)); err == nil {
|
||||
tender.ID = id
|
||||
}
|
||||
tender.ContractFolderID = item.ContractFolderID
|
||||
tender.NoticePublicationID = item.NoticePublicationID
|
||||
tender.RelatedNoticePublicationIDs = append([]string(nil), item.RelatedNoticePublicationIDs...)
|
||||
tender.Status = item.Status
|
||||
tender.PublicationDate = item.PublicationDate
|
||||
tender.SubmissionDeadline = item.SubmissionDeadline
|
||||
tender.TenderDeadline = item.TenderDeadline
|
||||
return tender
|
||||
}
|
||||
|
||||
func normalizePageCacheLanguages(defaultLanguage string, languages []string) []string {
|
||||
seen := make(map[string]struct{})
|
||||
out := make([]string, 0, len(languages)+1)
|
||||
|
||||
add := func(lang string) {
|
||||
lang = strings.ToLower(strings.TrimSpace(lang))
|
||||
if lang == "" {
|
||||
return
|
||||
}
|
||||
if _, ok := seen[lang]; ok {
|
||||
return
|
||||
}
|
||||
seen[lang] = struct{}{}
|
||||
out = append(out, lang)
|
||||
}
|
||||
|
||||
add(defaultLanguage)
|
||||
for _, lang := range languages {
|
||||
add(lang)
|
||||
}
|
||||
if len(out) == 0 {
|
||||
out = append(out, "en")
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// ParsePageCacheLanguagesConfig splits a comma-separated language list for cache prewarm.
|
||||
func ParsePageCacheLanguagesConfig(raw string) []string {
|
||||
if strings.TrimSpace(raw) == "" {
|
||||
return normalizePageCacheLanguages("en", nil)
|
||||
}
|
||||
return normalizePageCacheLanguages("", strings.Split(raw, ","))
|
||||
}
|
||||
|
||||
// RecommendedTendersPageCacheKeyPrefix is shared with company invalidation logic.
|
||||
const RecommendedTendersPageCacheKeyPrefix = recommendedTendersPageCacheKeyPrefix
|
||||
|
||||
// RecommendedTendersPageCacheKey builds the Redis key for a cached recommendation page.
|
||||
func RecommendedTendersPageCacheKey(companyID, language string) string {
|
||||
return recommendedTendersPageCacheKey(companyID, language)
|
||||
}
|
||||
@@ -0,0 +1,54 @@
|
||||
package tender
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"go.mongodb.org/mongo-driver/v2/bson"
|
||||
)
|
||||
|
||||
func TestFilterCachedRecommendedTendersAppliesActiveAndRejectedFilters(t *testing.T) {
|
||||
activeID, _ := bson.ObjectIDFromHex("507f1f77bcf86cd799439011")
|
||||
expiredID, _ := bson.ObjectIDFromHex("507f1f77bcf86cd799439012")
|
||||
|
||||
items := []TenderResponse{
|
||||
{
|
||||
ID: activeID.Hex(),
|
||||
Status: TenderStatusActive,
|
||||
PublicationDate: 1,
|
||||
SubmissionDeadline: 9999999999,
|
||||
ProcedureRef: "PROC_folder-a/notice-a",
|
||||
Rank: 1,
|
||||
},
|
||||
{
|
||||
ID: expiredID.Hex(),
|
||||
Status: TenderStatusExpired,
|
||||
ProcedureRef: "PROC_folder-b/notice-b",
|
||||
Rank: 2,
|
||||
},
|
||||
}
|
||||
|
||||
form := &SearchForm{
|
||||
OnlyActiveDeadlines: true,
|
||||
ExcludeRejectedTenders: true,
|
||||
Status: []string{string(TenderStatusActive)},
|
||||
}
|
||||
excluded := map[string]struct{}{expiredID.Hex(): {}}
|
||||
|
||||
got := filterCachedRecommendedTenders(items, form, nil, excluded, nil, 100)
|
||||
if len(got) != 1 {
|
||||
t.Fatalf("len = %d, want 1", len(got))
|
||||
}
|
||||
if got[0].ID != activeID.Hex() {
|
||||
t.Fatalf("got ID %q, want active tender", got[0].ID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestParsePageCacheLanguagesConfig(t *testing.T) {
|
||||
got := ParsePageCacheLanguagesConfig("en,fr, en")
|
||||
if len(got) != 2 {
|
||||
t.Fatalf("len = %d, want 2", len(got))
|
||||
}
|
||||
if got[0] != "en" || got[1] != "fr" {
|
||||
t.Fatalf("got %v, want [en fr]", got)
|
||||
}
|
||||
}
|
||||
@@ -362,13 +362,18 @@ func (r *tenderRepository) GetByProcedureReference(ctx context.Context, contract
|
||||
|
||||
const procedureReferenceLookupBatchSize = 250
|
||||
|
||||
// GetByProcedureReferences loads tenders for multiple AI procedure refs using indexed notice lookups.
|
||||
const contractFolderLookupBatchSize = 250
|
||||
|
||||
// GetByProcedureReferences loads tenders for multiple AI procedure refs.
|
||||
// It combines indexed notice lookups with contract-folder candidates so older notice
|
||||
// references still resolve after tender merges (same behavior as before the perf refactor).
|
||||
func (r *tenderRepository) GetByProcedureReferences(ctx context.Context, refs []ProcedureReference) (map[string]Tender, error) {
|
||||
out := make(map[string]Tender)
|
||||
if len(refs) == 0 {
|
||||
return out, nil
|
||||
}
|
||||
|
||||
folderIDs := make([]string, 0, len(refs))
|
||||
noticeIDs := make([]string, 0, len(refs))
|
||||
for _, ref := range refs {
|
||||
folder := strings.TrimSpace(ref.ContractFolderID)
|
||||
@@ -376,15 +381,12 @@ func (r *tenderRepository) GetByProcedureReferences(ctx context.Context, refs []
|
||||
if folder == "" || notice == "" {
|
||||
continue
|
||||
}
|
||||
folderIDs = append(folderIDs, folder)
|
||||
noticeIDs = append(noticeIDs, notice)
|
||||
}
|
||||
|
||||
uniqueNotices := uniqueNonEmptyTrimmedStrings(noticeIDs)
|
||||
if len(uniqueNotices) == 0 {
|
||||
return out, nil
|
||||
}
|
||||
|
||||
candidates := make([]Tender, 0, len(uniqueNotices))
|
||||
noticeCandidates := make([]Tender, 0, len(uniqueNotices))
|
||||
for start := 0; start < len(uniqueNotices); start += procedureReferenceLookupBatchSize {
|
||||
end := start + procedureReferenceLookupBatchSize
|
||||
if end > len(uniqueNotices) {
|
||||
@@ -395,12 +397,79 @@ func (r *tenderRepository) GetByProcedureReferences(ctx context.Context, refs []
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
candidates = append(candidates, batch...)
|
||||
noticeCandidates = append(noticeCandidates, batch...)
|
||||
}
|
||||
|
||||
folderCandidates, err := r.findTendersByContractFolderIDs(ctx, folderIDs)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
candidates := mergeTenderCandidates(noticeCandidates, folderCandidates)
|
||||
return mapProcedureReferencesToTenders(candidates, refs), nil
|
||||
}
|
||||
|
||||
func (r *tenderRepository) findTendersByContractFolderIDs(ctx context.Context, folderIDs []string) ([]Tender, error) {
|
||||
uniqueFolders := uniqueNonEmptyTrimmedStrings(folderIDs)
|
||||
if len(uniqueFolders) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
out := make([]Tender, 0, len(uniqueFolders))
|
||||
for start := 0; start < len(uniqueFolders); start += contractFolderLookupBatchSize {
|
||||
end := start + contractFolderLookupBatchSize
|
||||
if end > len(uniqueFolders) {
|
||||
end = len(uniqueFolders)
|
||||
}
|
||||
batch := uniqueFolders[start:end]
|
||||
|
||||
findLimit := len(batch) * 5
|
||||
if minLimit := len(batch) + 1; findLimit < minLimit {
|
||||
findLimit = minLimit
|
||||
}
|
||||
|
||||
filter := bson.M{"contract_folder_id": bson.M{"$in": batch}}
|
||||
pagination := orm.Pagination{
|
||||
Limit: findLimit,
|
||||
SortField: "updated_at",
|
||||
SortOrder: -1,
|
||||
SkipCount: true,
|
||||
}
|
||||
|
||||
result, err := r.ormRepo.FindAll(ctx, filter, pagination)
|
||||
if err != nil {
|
||||
r.logger.Error("Failed to get tenders by contract folder IDs", map[string]interface{}{
|
||||
"count": len(batch),
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, result.Items...)
|
||||
}
|
||||
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func mergeTenderCandidates(groups ...[]Tender) []Tender {
|
||||
seen := make(map[string]struct{})
|
||||
out := make([]Tender, 0)
|
||||
for _, group := range groups {
|
||||
for i := range group {
|
||||
tenderID := strings.TrimSpace(group[i].GetID())
|
||||
if tenderID == "" {
|
||||
out = append(out, group[i])
|
||||
continue
|
||||
}
|
||||
if _, exists := seen[tenderID]; exists {
|
||||
continue
|
||||
}
|
||||
seen[tenderID] = struct{}{}
|
||||
out = append(out, group[i])
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func (r *tenderRepository) findTendersByNoticePublicationIDs(ctx context.Context, noticeIDs []string) ([]Tender, error) {
|
||||
if len(noticeIDs) == 0 {
|
||||
return nil, nil
|
||||
@@ -413,7 +482,7 @@ func (r *tenderRepository) findTendersByNoticePublicationIDs(ctx context.Context
|
||||
},
|
||||
}
|
||||
pagination := orm.Pagination{
|
||||
Limit: len(noticeIDs) * 2,
|
||||
Limit: len(noticeIDs) * 3,
|
||||
SortField: "updated_at",
|
||||
SortOrder: -1,
|
||||
SkipCount: true,
|
||||
|
||||
@@ -16,6 +16,7 @@ import (
|
||||
"tm/pkg/ai_summarizer"
|
||||
"tm/pkg/logger"
|
||||
orm "tm/pkg/mongo"
|
||||
"tm/pkg/redis"
|
||||
"tm/pkg/response"
|
||||
|
||||
"golang.org/x/sync/errgroup"
|
||||
@@ -89,6 +90,9 @@ type tenderService struct {
|
||||
logger logger.Logger
|
||||
aiSummarizerClient AISummarizerClient
|
||||
aiSummarizerStorage AISummarizerStorage
|
||||
redisClient redis.Client
|
||||
pageCacheTTL time.Duration
|
||||
pageCacheLanguages []string
|
||||
defaultLanguage string
|
||||
}
|
||||
|
||||
@@ -111,6 +115,9 @@ func NewService(
|
||||
logger logger.Logger,
|
||||
aiClient AISummarizerClient,
|
||||
aiStorage AISummarizerStorage,
|
||||
redisClient redis.Client,
|
||||
recommendationPageCacheTTL time.Duration,
|
||||
recommendationPageCacheLanguages []string,
|
||||
defaultLanguage string,
|
||||
) Service {
|
||||
if strings.TrimSpace(defaultLanguage) == "" {
|
||||
@@ -125,6 +132,9 @@ func NewService(
|
||||
logger: logger,
|
||||
aiSummarizerClient: aiClient,
|
||||
aiSummarizerStorage: aiStorage,
|
||||
redisClient: redisClient,
|
||||
pageCacheTTL: recommendationPageCacheTTL,
|
||||
pageCacheLanguages: normalizePageCacheLanguages(defaultLanguage, recommendationPageCacheLanguages),
|
||||
defaultLanguage: strings.ToLower(defaultLanguage),
|
||||
}
|
||||
}
|
||||
@@ -967,6 +977,11 @@ func (s *tenderService) Recommend(ctx context.Context, form *SearchForm, paginat
|
||||
"company_ids": companyIDs,
|
||||
})
|
||||
|
||||
lang := s.pickResponseLanguage(form.Language)
|
||||
if result, ok := s.recommendFromPageCache(ctx, companyID, companyIDs, form, pagination, lang); ok {
|
||||
return result, nil
|
||||
}
|
||||
|
||||
var (
|
||||
recommendations []company.RecommendedTenderResponse
|
||||
err error
|
||||
@@ -983,6 +998,10 @@ func (s *tenderService) Recommend(ctx context.Context, form *SearchForm, paginat
|
||||
return s.emptySearchResponse(pagination), nil
|
||||
}
|
||||
|
||||
if len(companyIDs) == 0 && companyID != "" {
|
||||
s.ScheduleRefreshRecommendedTendersPageCache(companyID)
|
||||
}
|
||||
|
||||
excluded, excludedByCompany, err := s.loadRecommendationExclusions(ctx, form, companyID, companyIDs)
|
||||
if err != nil {
|
||||
if form.ExcludeRejectedTenders {
|
||||
@@ -995,7 +1014,6 @@ func (s *tenderService) Recommend(ctx context.Context, form *SearchForm, paginat
|
||||
return nil, err
|
||||
}
|
||||
|
||||
lang := s.pickResponseLanguage(form.Language)
|
||||
now := time.Now().Unix()
|
||||
pageResult, err := s.buildRecommendationPage(
|
||||
ctx,
|
||||
|
||||
Reference in New Issue
Block a user