Compare commits
10 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 10385e997b | |||
| adfd291245 | |||
| 26a593306d | |||
| c5e62a4061 | |||
| 81b0d94ba3 | |||
| 3e2700bc36 | |||
| aafae2a26f | |||
| 9d8ce12816 | |||
| b388af3518 | |||
| 932b0cf24e |
@@ -565,7 +565,6 @@ func (w *NoticeWorker) ToTender(n *notice.Notice) (*tender.Tender, error) {
|
||||
t.ProcurementProjectID = n.ProcurementProjectID
|
||||
t.SourceFileURL = n.SourceFileURL
|
||||
t.SourceFileName = n.SourceFileName
|
||||
t.ContentXML = n.ContentXML
|
||||
// Preserve document-scraping/summarization flags on the tender (managed by
|
||||
// downstream workers), but refresh the notice-side processing metadata.
|
||||
t.ProcessingMetadata.ScrapedAt = n.ProcessingMetadata.ScrapedAt
|
||||
|
||||
@@ -441,6 +441,22 @@ func (r *repository) ClosingSoon(ctx context.Context, limit int, windowSec int64
|
||||
windowEnd := now + windowSec
|
||||
|
||||
pipeline := mongo.Pipeline{
|
||||
{{Key: "$match", Value: bson.M{
|
||||
"$or": bson.A{
|
||||
bson.M{
|
||||
"$and": bson.A{
|
||||
bson.M{"submission_deadline": bson.M{"$gt": 0}},
|
||||
deadlineWithinWindowMatch("submission_deadline", now, windowEnd),
|
||||
},
|
||||
},
|
||||
bson.M{
|
||||
"$and": bson.A{
|
||||
noSubmissionDeadlineClause(),
|
||||
deadlineWithinWindowMatch("tender_deadline", now, windowEnd),
|
||||
},
|
||||
},
|
||||
},
|
||||
}}},
|
||||
{{Key: "$addFields", Value: bson.M{"effective_deadline": effectiveDeadlineExpr()}}},
|
||||
{{Key: "$match", Value: bson.M{
|
||||
"effective_deadline": bson.M{"$gt": now, "$lte": windowEnd},
|
||||
@@ -580,6 +596,26 @@ func trendTimestampField(metric string) string {
|
||||
}
|
||||
}
|
||||
|
||||
// deadlineWithinWindowMatch matches raw deadline fields stored as Unix seconds or
|
||||
// milliseconds. normalizeTimestampExpr treats values > 1e12 as ms; indexed pre-filters
|
||||
// must accept both encodings to stay equivalent to filtering on normalized deadlines.
|
||||
func deadlineWithinWindowMatch(field string, now, windowEnd int64) bson.M {
|
||||
return bson.M{
|
||||
"$or": bson.A{
|
||||
bson.M{field: bson.M{"$gt": now, "$lte": windowEnd}},
|
||||
bson.M{field: bson.M{"$gt": now * 1000, "$lte": windowEnd * 1000}},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func noSubmissionDeadlineClause() bson.M {
|
||||
return bson.M{"$or": bson.A{
|
||||
bson.M{"submission_deadline": bson.M{"$exists": false}},
|
||||
bson.M{"submission_deadline": nil},
|
||||
bson.M{"submission_deadline": bson.M{"$lte": 0}},
|
||||
}}
|
||||
}
|
||||
|
||||
// effectiveDeadlineExpr is used for closing-soon (submission first, per dashboard-api.md).
|
||||
func effectiveDeadlineExpr() bson.M {
|
||||
return normalizeDeadlineFieldExpr(
|
||||
|
||||
@@ -42,3 +42,30 @@ func TestFacetCurrencyValueDecodesBSOND(t *testing.T) {
|
||||
t.Fatalf("expected 12345.67, got %f", total)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDeadlineWithinWindowMatchCoversSecondsAndMilliseconds(t *testing.T) {
|
||||
now := int64(1_700_000_000)
|
||||
windowEnd := now + 86_400
|
||||
|
||||
match := deadlineWithinWindowMatch("submission_deadline", now, windowEnd)
|
||||
orClause, ok := match["$or"].(bson.A)
|
||||
if !ok || len(orClause) != 2 {
|
||||
t.Fatalf("expected two range branches, got %#v", match)
|
||||
}
|
||||
|
||||
secRange, ok := orClause[0].(bson.M)["submission_deadline"].(bson.M)
|
||||
if !ok {
|
||||
t.Fatalf("expected seconds range on submission_deadline, got %#v", orClause[0])
|
||||
}
|
||||
if secRange["$gt"] != now || secRange["$lte"] != windowEnd {
|
||||
t.Fatalf("unexpected seconds range: %#v", secRange)
|
||||
}
|
||||
|
||||
msRange, ok := orClause[1].(bson.M)["submission_deadline"].(bson.M)
|
||||
if !ok {
|
||||
t.Fatalf("expected milliseconds range on submission_deadline, got %#v", orClause[1])
|
||||
}
|
||||
if msRange["$gt"] != now*1000 || msRange["$lte"] != windowEnd*1000 {
|
||||
t.Fatalf("unexpected milliseconds range: %#v", msRange)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -60,6 +60,11 @@ type service struct {
|
||||
statisticsMu sync.Mutex
|
||||
statistics map[int]cacheEntry[*StatisticsReportResponse]
|
||||
statisticsGroup singleflight.Group
|
||||
trendCache *widgetCache[*TrendResponse]
|
||||
countriesCache *widgetCache[*CountriesResponse]
|
||||
noticeTypesCache *widgetCache[*NoticeTypesResponse]
|
||||
closingSoonCache *widgetCache[*ClosingSoonResponse]
|
||||
recentCache *widgetCache[*RecentResponse]
|
||||
}
|
||||
|
||||
// NewService creates a dashboard service.
|
||||
@@ -70,9 +75,15 @@ func NewService(repo Repository, log logger.Logger, redisClient redis.Client) Se
|
||||
redis: redisClient,
|
||||
summary: make(map[string]cacheEntry[*SummaryResponse]),
|
||||
statistics: make(map[int]cacheEntry[*StatisticsReportResponse]),
|
||||
trendCache: newWidgetCache[*TrendResponse](redisClient, log, "trend"),
|
||||
countriesCache: newWidgetCache[*CountriesResponse](redisClient, log, "countries"),
|
||||
noticeTypesCache: newWidgetCache[*NoticeTypesResponse](redisClient, log, "notice-types"),
|
||||
closingSoonCache: newWidgetCache[*ClosingSoonResponse](redisClient, log, "closing-soon"),
|
||||
recentCache: newWidgetCache[*RecentResponse](redisClient, log, "recent"),
|
||||
}
|
||||
go s.warmStatisticsCache()
|
||||
go s.warmSummaryCache()
|
||||
go s.warmWidgetCaches()
|
||||
return s
|
||||
}
|
||||
|
||||
@@ -129,6 +140,9 @@ func (s *service) Summary(ctx context.Context, query SummaryQuery) (*SummaryResp
|
||||
func (s *service) Trend(ctx context.Context, query TrendQuery) (*TrendResponse, error) {
|
||||
days := trendDays(query.Days)
|
||||
metric := normalizeTrendMetric(query.Metric)
|
||||
cacheKey := fmt.Sprintf("%d:%s", days, metric)
|
||||
|
||||
return s.trendCache.Get(ctx, cacheKey, func(loadCtx context.Context) (*TrendResponse, error) {
|
||||
startUnix := trendStartUnix(days)
|
||||
|
||||
s.logger.Info("Fetching dashboard trend", map[string]interface{}{
|
||||
@@ -136,7 +150,7 @@ func (s *service) Trend(ctx context.Context, query TrendQuery) (*TrendResponse,
|
||||
"metric": metric,
|
||||
})
|
||||
|
||||
counts, err := s.repo.Trend(ctx, days, metric, startUnix)
|
||||
counts, err := s.repo.Trend(loadCtx, days, metric, startUnix)
|
||||
if err != nil {
|
||||
s.logger.Error("Failed to fetch dashboard trend", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
@@ -144,23 +158,24 @@ func (s *service) Trend(ctx context.Context, query TrendQuery) (*TrendResponse,
|
||||
return nil, fmt.Errorf("dashboard trend: %w", err)
|
||||
}
|
||||
|
||||
series := fillTrendSeries(days, counts)
|
||||
|
||||
return &TrendResponse{
|
||||
Metric: metric,
|
||||
Days: days,
|
||||
Series: series,
|
||||
Series: fillTrendSeries(days, counts),
|
||||
}, nil
|
||||
})
|
||||
}
|
||||
|
||||
func (s *service) Countries(ctx context.Context, query CountriesQuery) (*CountriesResponse, error) {
|
||||
limit := countriesLimit(query.Limit)
|
||||
cacheKey := strconv.Itoa(limit)
|
||||
|
||||
return s.countriesCache.Get(ctx, cacheKey, func(loadCtx context.Context) (*CountriesResponse, error) {
|
||||
s.logger.Info("Fetching dashboard countries", map[string]interface{}{
|
||||
"limit": limit,
|
||||
})
|
||||
|
||||
out, err := s.repo.Countries(ctx, limit)
|
||||
out, err := s.repo.Countries(loadCtx, limit)
|
||||
if err != nil {
|
||||
s.logger.Error("Failed to fetch dashboard countries", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
@@ -169,12 +184,14 @@ func (s *service) Countries(ctx context.Context, query CountriesQuery) (*Countri
|
||||
}
|
||||
|
||||
return out, nil
|
||||
})
|
||||
}
|
||||
|
||||
func (s *service) NoticeTypes(ctx context.Context) (*NoticeTypesResponse, error) {
|
||||
return s.noticeTypesCache.Get(ctx, "all", func(loadCtx context.Context) (*NoticeTypesResponse, error) {
|
||||
s.logger.Info("Fetching dashboard notice types", nil)
|
||||
|
||||
out, err := s.repo.NoticeTypes(ctx)
|
||||
out, err := s.repo.NoticeTypes(loadCtx)
|
||||
if err != nil {
|
||||
s.logger.Error("Failed to fetch dashboard notice types", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
@@ -183,19 +200,22 @@ func (s *service) NoticeTypes(ctx context.Context) (*NoticeTypesResponse, error)
|
||||
}
|
||||
|
||||
return out, nil
|
||||
})
|
||||
}
|
||||
|
||||
func (s *service) ClosingSoon(ctx context.Context, query ClosingSoonQuery) (*ClosingSoonResponse, error) {
|
||||
limit := listLimit(query.Limit)
|
||||
windowHours := closingWindowHours(query.Window)
|
||||
windowSec := int64(windowHours) * 3600
|
||||
cacheKey := fmt.Sprintf("%d:%d", limit, windowSec)
|
||||
|
||||
return s.closingSoonCache.Get(ctx, cacheKey, func(loadCtx context.Context) (*ClosingSoonResponse, error) {
|
||||
s.logger.Info("Fetching dashboard closing soon", map[string]interface{}{
|
||||
"limit": limit,
|
||||
"window": windowHours,
|
||||
})
|
||||
|
||||
items, err := s.repo.ClosingSoon(ctx, limit, windowSec)
|
||||
items, err := s.repo.ClosingSoon(loadCtx, limit, windowSec)
|
||||
if err != nil {
|
||||
s.logger.Error("Failed to fetch dashboard closing soon", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
@@ -204,16 +224,18 @@ func (s *service) ClosingSoon(ctx context.Context, query ClosingSoonQuery) (*Clo
|
||||
}
|
||||
|
||||
return &ClosingSoonResponse{Items: items}, nil
|
||||
})
|
||||
}
|
||||
|
||||
func (s *service) Recent(ctx context.Context, query RecentQuery) (*RecentResponse, error) {
|
||||
limit := listLimit(query.Limit)
|
||||
|
||||
cursor := strings.TrimSpace(query.Cursor)
|
||||
if cursor != "" {
|
||||
s.logger.Info("Fetching dashboard recent tenders", map[string]interface{}{
|
||||
"limit": limit,
|
||||
})
|
||||
|
||||
items, nextCursor, err := s.repo.Recent(ctx, limit, strings.TrimSpace(query.Cursor))
|
||||
items, nextCursor, err := s.repo.Recent(ctx, limit, cursor)
|
||||
if err != nil {
|
||||
s.logger.Error("Failed to fetch dashboard recent tenders", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
@@ -225,6 +247,27 @@ func (s *service) Recent(ctx context.Context, query RecentQuery) (*RecentRespons
|
||||
Items: items,
|
||||
NextCursor: nextCursor,
|
||||
}, nil
|
||||
}
|
||||
|
||||
cacheKey := strconv.Itoa(limit)
|
||||
return s.recentCache.Get(ctx, cacheKey, func(loadCtx context.Context) (*RecentResponse, error) {
|
||||
s.logger.Info("Fetching dashboard recent tenders", map[string]interface{}{
|
||||
"limit": limit,
|
||||
})
|
||||
|
||||
items, nextCursor, err := s.repo.Recent(loadCtx, limit, "")
|
||||
if err != nil {
|
||||
s.logger.Error("Failed to fetch dashboard recent tenders", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil, fmt.Errorf("dashboard recent: %w", err)
|
||||
}
|
||||
|
||||
return &RecentResponse{
|
||||
Items: items,
|
||||
NextCursor: nextCursor,
|
||||
}, nil
|
||||
})
|
||||
}
|
||||
|
||||
func (s *service) Statistics(ctx context.Context, query StatisticsQuery) (*StatisticsReportResponse, error) {
|
||||
@@ -339,6 +382,80 @@ func (s *service) loadStatistics(ctx context.Context, days int) (*StatisticsRepo
|
||||
return out.(*StatisticsReportResponse), nil
|
||||
}
|
||||
|
||||
func (s *service) warmWidgetCaches() {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
|
||||
defaultWindowSec := int64(defaultClosingWindowHours) * 3600
|
||||
warmed := 0
|
||||
|
||||
if s.trendCache.WarmFromRedis(ctx, fmt.Sprintf("%d:%s", defaultTrendDays, "created")) {
|
||||
warmed++
|
||||
}
|
||||
if s.countriesCache.WarmFromRedis(ctx, strconv.Itoa(defaultCountriesLimit)) {
|
||||
warmed++
|
||||
}
|
||||
if s.noticeTypesCache.WarmFromRedis(ctx, "all") {
|
||||
warmed++
|
||||
}
|
||||
if s.closingSoonCache.WarmFromRedis(ctx, fmt.Sprintf("%d:%d", defaultListLimit, defaultWindowSec)) {
|
||||
warmed++
|
||||
}
|
||||
if s.recentCache.WarmFromRedis(ctx, strconv.Itoa(defaultListLimit)) {
|
||||
warmed++
|
||||
}
|
||||
|
||||
if warmed > 0 {
|
||||
s.logger.Info("Dashboard widget caches warmed from Redis", map[string]interface{}{
|
||||
"warmed": warmed,
|
||||
})
|
||||
}
|
||||
|
||||
go s.trendCache.Reload(fmt.Sprintf("%d:%s", defaultTrendDays, "created"), func(loadCtx context.Context) (*TrendResponse, error) {
|
||||
return s.loadTrend(loadCtx, TrendQuery{Days: defaultTrendDays, Metric: "created"})
|
||||
})
|
||||
go s.countriesCache.Reload(strconv.Itoa(defaultCountriesLimit), func(loadCtx context.Context) (*CountriesResponse, error) {
|
||||
return s.repo.Countries(loadCtx, defaultCountriesLimit)
|
||||
})
|
||||
go s.noticeTypesCache.Reload("all", func(loadCtx context.Context) (*NoticeTypesResponse, error) {
|
||||
return s.repo.NoticeTypes(loadCtx)
|
||||
})
|
||||
go s.closingSoonCache.Reload(fmt.Sprintf("%d:%d", defaultListLimit, defaultWindowSec), func(loadCtx context.Context) (*ClosingSoonResponse, error) {
|
||||
items, err := s.repo.ClosingSoon(loadCtx, defaultListLimit, defaultWindowSec)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &ClosingSoonResponse{Items: items}, nil
|
||||
})
|
||||
go s.recentCache.Reload(strconv.Itoa(defaultListLimit), func(loadCtx context.Context) (*RecentResponse, error) {
|
||||
items, nextCursor, err := s.repo.Recent(loadCtx, defaultListLimit, "")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &RecentResponse{
|
||||
Items: items,
|
||||
NextCursor: nextCursor,
|
||||
}, nil
|
||||
})
|
||||
}
|
||||
|
||||
func (s *service) loadTrend(ctx context.Context, query TrendQuery) (*TrendResponse, error) {
|
||||
days := trendDays(query.Days)
|
||||
metric := normalizeTrendMetric(query.Metric)
|
||||
startUnix := trendStartUnix(days)
|
||||
|
||||
counts, err := s.repo.Trend(ctx, days, metric, startUnix)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &TrendResponse{
|
||||
Metric: metric,
|
||||
Days: days,
|
||||
Series: fillTrendSeries(days, counts),
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *service) warmSummaryCache() {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
|
||||
@@ -0,0 +1,230 @@
|
||||
package dashboard
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
"tm/pkg/logger"
|
||||
"tm/pkg/redis"
|
||||
|
||||
goredis "github.com/redis/go-redis/v9"
|
||||
"golang.org/x/sync/singleflight"
|
||||
)
|
||||
|
||||
const (
|
||||
// widgetCacheTTL is the fresh window before a background reload is triggered.
|
||||
widgetCacheTTL = 60 * time.Second
|
||||
// widgetStaleGraceTTL is how long stale entries may be served while reloading.
|
||||
// Dashboard widgets have no write-side invalidation; expect up to this lag after
|
||||
// data changes (e.g. a new tender may not appear in cached widgets immediately).
|
||||
widgetStaleGraceTTL = 5 * time.Minute
|
||||
widgetReloadTimeout = 2 * time.Minute
|
||||
)
|
||||
|
||||
type widgetLoader[T any] func(ctx context.Context) (T, error)
|
||||
|
||||
type widgetCache[T any] struct {
|
||||
mu sync.Mutex
|
||||
entries map[string]cacheEntry[T]
|
||||
group singleflight.Group
|
||||
redis redis.Client
|
||||
logger logger.Logger
|
||||
keyPrefix string
|
||||
ttl time.Duration
|
||||
staleGrace time.Duration
|
||||
}
|
||||
|
||||
func newWidgetCache[T any](redisClient redis.Client, log logger.Logger, keyPrefix string) *widgetCache[T] {
|
||||
return &widgetCache[T]{
|
||||
entries: make(map[string]cacheEntry[T]),
|
||||
redis: redisClient,
|
||||
logger: log,
|
||||
keyPrefix: keyPrefix,
|
||||
ttl: widgetCacheTTL,
|
||||
staleGrace: widgetStaleGraceTTL,
|
||||
}
|
||||
}
|
||||
|
||||
func (c *widgetCache[T]) Get(ctx context.Context, key string, load widgetLoader[T]) (T, error) {
|
||||
var zero T
|
||||
|
||||
// Returned values are shared across concurrent callers via singleflight and stale
|
||||
// hits. Callers must treat them as read-only through JSON serialization.
|
||||
if value, ok := c.fresh(key); ok {
|
||||
return value, nil
|
||||
}
|
||||
if value, ok := c.stale(key); ok {
|
||||
go c.Reload(key, load)
|
||||
return value, nil
|
||||
}
|
||||
if value, ok := c.fromRedis(ctx, key); ok {
|
||||
c.store(key, value)
|
||||
go c.Reload(key, load)
|
||||
return value, nil
|
||||
}
|
||||
|
||||
out, err, _ := c.group.Do(key, func() (interface{}, error) {
|
||||
if value, ok := c.fresh(key); ok {
|
||||
return value, nil
|
||||
}
|
||||
|
||||
result, err := load(context.WithoutCancel(ctx))
|
||||
if err != nil {
|
||||
return zero, err
|
||||
}
|
||||
|
||||
c.store(key, result)
|
||||
c.storeRedis(context.Background(), key, result)
|
||||
return result, nil
|
||||
})
|
||||
if err != nil {
|
||||
return zero, err
|
||||
}
|
||||
|
||||
return out.(T), nil
|
||||
}
|
||||
|
||||
func (c *widgetCache[T]) Reload(key string, load widgetLoader[T]) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), widgetReloadTimeout)
|
||||
defer cancel()
|
||||
|
||||
_, _, _ = c.group.Do("reload:"+key, func() (interface{}, error) {
|
||||
result, err := load(ctx)
|
||||
if err != nil {
|
||||
if c.logger != nil {
|
||||
c.logger.Warn("Failed to refresh dashboard widget cache", map[string]interface{}{
|
||||
"widget": c.keyPrefix,
|
||||
"key": key,
|
||||
"error": err.Error(),
|
||||
})
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
|
||||
c.store(key, result)
|
||||
c.storeRedis(context.Background(), key, result)
|
||||
return result, nil
|
||||
})
|
||||
}
|
||||
|
||||
func (c *widgetCache[T]) WarmFromRedis(ctx context.Context, key string) bool {
|
||||
value, ok := c.fromRedis(ctx, key)
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
c.store(key, value)
|
||||
return true
|
||||
}
|
||||
|
||||
func (c *widgetCache[T]) fresh(key string) (T, bool) {
|
||||
var zero T
|
||||
now := time.Now()
|
||||
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
entry, ok := c.entries[key]
|
||||
if !ok || now.After(entry.expiresAt) {
|
||||
return zero, false
|
||||
}
|
||||
|
||||
return entry.value, true
|
||||
}
|
||||
|
||||
func (c *widgetCache[T]) stale(key string) (T, bool) {
|
||||
var zero T
|
||||
now := time.Now()
|
||||
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
entry, ok := c.entries[key]
|
||||
if !ok || now.After(entry.staleUntil) {
|
||||
if ok {
|
||||
delete(c.entries, key)
|
||||
}
|
||||
return zero, false
|
||||
}
|
||||
|
||||
return entry.value, true
|
||||
}
|
||||
|
||||
func (c *widgetCache[T]) store(key string, value T) {
|
||||
now := time.Now()
|
||||
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
c.entries[key] = cacheEntry[T]{
|
||||
expiresAt: now.Add(c.ttl),
|
||||
staleUntil: now.Add(c.ttl + c.staleGrace),
|
||||
value: value,
|
||||
}
|
||||
}
|
||||
|
||||
func (c *widgetCache[T]) redisKey(key string) string {
|
||||
return fmt.Sprintf("dashboard:%s:%s", c.keyPrefix, key)
|
||||
}
|
||||
|
||||
func (c *widgetCache[T]) fromRedis(ctx context.Context, key string) (T, bool) {
|
||||
var zero T
|
||||
if c.redis == nil {
|
||||
return zero, false
|
||||
}
|
||||
|
||||
raw, err := c.redis.Get(ctx, c.redisKey(key))
|
||||
if err != nil {
|
||||
if err != goredis.Nil && c.logger != nil {
|
||||
c.logger.Warn("Failed to read dashboard widget cache from Redis", map[string]interface{}{
|
||||
"widget": c.keyPrefix,
|
||||
"key": key,
|
||||
"error": err.Error(),
|
||||
})
|
||||
}
|
||||
return zero, false
|
||||
}
|
||||
|
||||
var value T
|
||||
if err := json.Unmarshal([]byte(raw), &value); err != nil {
|
||||
if c.logger != nil {
|
||||
c.logger.Warn("Failed to decode dashboard widget cache from Redis", map[string]interface{}{
|
||||
"widget": c.keyPrefix,
|
||||
"key": key,
|
||||
"error": err.Error(),
|
||||
})
|
||||
}
|
||||
_ = c.redis.Del(ctx, c.redisKey(key))
|
||||
return zero, false
|
||||
}
|
||||
|
||||
return value, true
|
||||
}
|
||||
|
||||
func (c *widgetCache[T]) storeRedis(ctx context.Context, key string, value T) {
|
||||
if c.redis == nil {
|
||||
return
|
||||
}
|
||||
|
||||
encoded, err := json.Marshal(value)
|
||||
if err != nil {
|
||||
if c.logger != nil {
|
||||
c.logger.Warn("Failed to encode dashboard widget cache for Redis", map[string]interface{}{
|
||||
"widget": c.keyPrefix,
|
||||
"key": key,
|
||||
"error": err.Error(),
|
||||
})
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
ttl := c.ttl + c.staleGrace
|
||||
if err := c.redis.Set(ctx, c.redisKey(key), string(encoded), ttl); err != nil && c.logger != nil {
|
||||
c.logger.Warn("Failed to store dashboard widget cache in Redis", map[string]interface{}{
|
||||
"widget": c.keyPrefix,
|
||||
"key": key,
|
||||
"error": err.Error(),
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -28,8 +28,26 @@ const (
|
||||
DeliveryChannelInApp DeliveryChannel = "in_app"
|
||||
)
|
||||
|
||||
// EventTypeInApp identifies notifications stored for the in-app notification center.
|
||||
const EventTypeInApp = "in_app"
|
||||
// Persisted event types for stored notifications (admin list / notification center).
|
||||
const (
|
||||
EventTypeInApp = "in_app"
|
||||
EventTypeEmail = "email"
|
||||
EventTypePush = "push"
|
||||
)
|
||||
|
||||
// persistedEventType maps a delivery channel to the event_type stored in MongoDB.
|
||||
func persistedEventType(channel DeliveryChannel) (string, bool) {
|
||||
switch channel {
|
||||
case DeliveryChannelEmail:
|
||||
return EventTypeEmail, true
|
||||
case DeliveryChannelPush:
|
||||
return EventTypePush, true
|
||||
case DeliveryChannelInApp:
|
||||
return EventTypeInApp, true
|
||||
default:
|
||||
return "", false
|
||||
}
|
||||
}
|
||||
|
||||
// DeliveryStatus represents the delivery status of notification
|
||||
type NotificationStatus string
|
||||
|
||||
@@ -99,13 +99,27 @@ func (s *notificationService) Send(ctx context.Context, req *NotificationRequest
|
||||
}
|
||||
|
||||
func (s *notificationService) sendToRecipient(ctx context.Context, recipient notificationRecipient, req *NotificationRequest, link, image string) {
|
||||
if err := s.persistNotification(ctx, recipient, req, link, image); err != nil {
|
||||
// Always persist an in-app record so every send appears in the notification center.
|
||||
if err := s.persistNotification(ctx, recipient, req, link, image, DeliveryChannelInApp); err != nil {
|
||||
s.logger.Error("Failed to persist in-app notification", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
"user_id": recipient.UserID,
|
||||
})
|
||||
}
|
||||
|
||||
for _, channel := range req.Channels {
|
||||
if channel == DeliveryChannelInApp {
|
||||
continue
|
||||
}
|
||||
if err := s.persistNotification(ctx, recipient, req, link, image, channel); err != nil {
|
||||
s.logger.Error("Failed to persist notification", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
"user_id": recipient.UserID,
|
||||
"channel": channel,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
metadata := map[string]any{
|
||||
"tender": strings.TrimSpace(req.Tender),
|
||||
}
|
||||
@@ -149,15 +163,26 @@ func (s *notificationService) sendToRecipient(ctx context.Context, recipient not
|
||||
}
|
||||
}
|
||||
|
||||
func (s *notificationService) persistNotification(ctx context.Context, recipient notificationRecipient, req *NotificationRequest, link, image string) error {
|
||||
methods := map[string]string{
|
||||
"in_app": "true",
|
||||
func (s *notificationService) persistNotification(ctx context.Context, recipient notificationRecipient, req *NotificationRequest, link, image string, channel DeliveryChannel) error {
|
||||
eventType, ok := persistedEventType(channel)
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
|
||||
methods := map[string]string{}
|
||||
switch channel {
|
||||
case DeliveryChannelEmail:
|
||||
if recipient.Email == "" {
|
||||
return nil
|
||||
}
|
||||
if slices.Contains(req.Channels, DeliveryChannelEmail) && recipient.Email != "" {
|
||||
methods["email"] = recipient.Email
|
||||
case DeliveryChannelPush:
|
||||
if len(recipient.DeviceTokens) == 0 {
|
||||
return nil
|
||||
}
|
||||
if slices.Contains(req.Channels, DeliveryChannelPush) && len(recipient.DeviceTokens) > 0 {
|
||||
methods["push"] = recipient.DeviceTokens[0]
|
||||
case DeliveryChannelInApp:
|
||||
methods["in_app"] = "true"
|
||||
}
|
||||
|
||||
status := string(DeliveryStatusSent)
|
||||
@@ -173,7 +198,7 @@ func (s *notificationService) persistNotification(ctx context.Context, recipient
|
||||
Image: image,
|
||||
Priority: string(req.Priority),
|
||||
Type: string(req.Type),
|
||||
EventType: EventTypeInApp,
|
||||
EventType: eventType,
|
||||
Status: status,
|
||||
Methods: methods,
|
||||
Metadata: map[string]any{
|
||||
|
||||
@@ -88,7 +88,6 @@ type Tender struct {
|
||||
Source TenderSource `bson:"source" json:"source"`
|
||||
SourceFileURL string `bson:"source_file_url" json:"source_file_url"`
|
||||
SourceFileName string `bson:"source_file_name" json:"source_file_name"`
|
||||
ContentXML string `bson:"content_xml,omitempty" json:"content_xml,omitempty"`
|
||||
ProcessingMetadata ProcessingMetadata `bson:"processing_metadata" json:"processing_metadata"`
|
||||
// AIOverallSummary optional denormalized overall summary for search/list (e.g. synced from AI pipeline storage).
|
||||
AIOverallSummary string `bson:"ai_overall_summary,omitempty" json:"ai_overall_summary,omitempty"`
|
||||
|
||||
@@ -70,6 +70,79 @@ type SearchForm struct {
|
||||
ExcludeRejectedTenders bool `query:"-" valid:"optional"`
|
||||
}
|
||||
|
||||
// HasListFilters reports whether the form applies any MongoDB list filters.
|
||||
func (f *SearchForm) HasListFilters() bool {
|
||||
if f == nil {
|
||||
return false
|
||||
}
|
||||
if f.DocumentsScraped || f.OnlyActiveDeadlines || f.ExcludeRejectedTenders {
|
||||
return true
|
||||
}
|
||||
if f.Search != nil && strings.TrimSpace(*f.Search) != "" {
|
||||
return true
|
||||
}
|
||||
if f.Title != nil && strings.TrimSpace(*f.Title) != "" {
|
||||
return true
|
||||
}
|
||||
if f.Description != nil && strings.TrimSpace(*f.Description) != "" {
|
||||
return true
|
||||
}
|
||||
if f.NoticeType != nil && strings.TrimSpace(*f.NoticeType) != "" {
|
||||
return true
|
||||
}
|
||||
if len(f.NoticeTypes) > 0 || len(f.FormTypes) > 0 {
|
||||
return true
|
||||
}
|
||||
if f.ProcurementType != nil && strings.TrimSpace(*f.ProcurementType) != "" {
|
||||
return true
|
||||
}
|
||||
if f.Country != nil && strings.TrimSpace(*f.Country) != "" {
|
||||
return true
|
||||
}
|
||||
if f.CountryCode != nil && strings.TrimSpace(*f.CountryCode) != "" {
|
||||
return true
|
||||
}
|
||||
if len(f.CountryCodes) > 0 || len(f.RegionCodes) > 0 {
|
||||
return true
|
||||
}
|
||||
if f.MainClassification != nil && strings.TrimSpace(*f.MainClassification) != "" {
|
||||
return true
|
||||
}
|
||||
if len(f.Classifications) > 0 || len(f.CpvCodes) > 0 {
|
||||
return true
|
||||
}
|
||||
if f.MinEstimatedValue != nil || f.MaxEstimatedValue != nil || f.Currency != "" {
|
||||
return true
|
||||
}
|
||||
if f.CreatedAt != nil || f.CreatedAtFrom != nil || f.CreatedAtTo != nil {
|
||||
return true
|
||||
}
|
||||
if f.TenderDeadline != nil || f.TenderDeadlineFrom != nil || f.TenderDeadlineTo != nil ||
|
||||
f.DeadlineFrom != nil || f.DeadlineTo != nil {
|
||||
return true
|
||||
}
|
||||
if f.PublicationDate != nil || f.PublicationDateFrom != nil || f.PublicationDateTo != nil {
|
||||
return true
|
||||
}
|
||||
if f.SubmissionDeadline != nil || f.SubmissionDateAliasFrom != nil || f.SubmissionDateAliasTo != nil ||
|
||||
f.SubmissionDateFrom != nil || f.SubmissionDateTo != nil {
|
||||
return true
|
||||
}
|
||||
if len(f.Status) > 0 || len(f.Source) > 0 || len(f.Languages) > 0 {
|
||||
return true
|
||||
}
|
||||
if f.BuyerOrganizationID != nil && strings.TrimSpace(*f.BuyerOrganizationID) != "" {
|
||||
return true
|
||||
}
|
||||
if f.NoticePublicationID != nil && strings.TrimSpace(*f.NoticePublicationID) != "" {
|
||||
return true
|
||||
}
|
||||
if len(f.ContractFolderIDsWithDocuments) > 0 {
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// SearchResponse represents the response for listing tenders
|
||||
type SearchResponse struct {
|
||||
Tenders []TenderResponse `json:"tenders"`
|
||||
|
||||
@@ -27,7 +27,6 @@ type TenderRepository interface {
|
||||
// GetLatestByContractFolderIDs returns the most recently updated tender per distinct contract_folder_id.
|
||||
GetLatestByContractFolderIDs(ctx context.Context, contractFolderIDs []string) (map[string]*Tender, error)
|
||||
GetByProcurementProjectID(ctx context.Context, procurementProjectID string) (*Tender, error)
|
||||
FindTendersWithContentXML(ctx context.Context, limit, skip int) ([]Tender, error)
|
||||
GetTenderCountByCountry(ctx context.Context) (map[string]int64, error)
|
||||
GetTenderCountByType(ctx context.Context) (map[string]int64, error)
|
||||
GetTenderCountByClassification(ctx context.Context) (map[string]int64, error)
|
||||
@@ -130,6 +129,15 @@ func NewRepository(mongoManager *orm.ConnectionManager, logger logger.Logger) Te
|
||||
*orm.NewIndex("main_classification_idx", bson.D{{Key: "main_classification", Value: 1}}),
|
||||
*orm.NewIndex("publication_date_idx", bson.D{{Key: "publication_date", Value: -1}}),
|
||||
*orm.NewIndex("tender_deadline_idx", bson.D{{Key: "tender_deadline", Value: 1}}),
|
||||
*orm.NewIndex("submission_deadline_idx", bson.D{{Key: "submission_deadline", Value: 1}}),
|
||||
*orm.NewIndex("closing_soon_submission_idx", bson.D{
|
||||
{Key: "submission_deadline", Value: 1},
|
||||
{Key: "status", Value: 1},
|
||||
}),
|
||||
*orm.NewIndex("closing_soon_tender_deadline_idx", bson.D{
|
||||
{Key: "tender_deadline", Value: 1},
|
||||
{Key: "status", Value: 1},
|
||||
}),
|
||||
*orm.NewIndex("estimated_value_idx", bson.D{{Key: "estimated_value", Value: -1}}),
|
||||
*orm.NewIndex("created_at_idx", bson.D{{Key: "created_at", Value: -1}}),
|
||||
// Stable keyset pagination for default admin list (created_at desc).
|
||||
@@ -172,6 +180,13 @@ func NewRepository(mongoManager *orm.ConnectionManager, logger logger.Logger) Te
|
||||
{Key: "processing_metadata.documents_scraped", Value: 1},
|
||||
{Key: "processing_metadata.documents_scraped_at", Value: -1},
|
||||
}),
|
||||
// Panel list: documents_scraped=true sorted by created_at desc (keyset pagination).
|
||||
*orm.NewIndex("documents_scraped_created_at_id_idx", bson.D{
|
||||
{Key: "created_at", Value: -1},
|
||||
{Key: "_id", Value: -1},
|
||||
}).WithPartialFilterExpression(bson.M{
|
||||
"processing_metadata.documents_scraped": true,
|
||||
}),
|
||||
*orm.NewIndex("scraped_documents_scraped_at_idx", bson.D{
|
||||
{Key: "scraped_documents.scraped_at", Value: 1},
|
||||
}),
|
||||
@@ -660,27 +675,6 @@ func (r *tenderRepository) GetByProcurementProjectID(ctx context.Context, procur
|
||||
return &result.Items[0], nil
|
||||
}
|
||||
|
||||
// FindTendersWithContentXML returns tenders that still have TED XML (for backfill after notices were deleted).
|
||||
func (r *tenderRepository) FindTendersWithContentXML(ctx context.Context, limit, skip int) ([]Tender, error) {
|
||||
filter := bson.M{"content_xml": bson.M{"$exists": true, "$ne": ""}}
|
||||
pagination := orm.Pagination{
|
||||
Limit: limit,
|
||||
Skip: skip,
|
||||
SortField: "_id",
|
||||
SortOrder: 1,
|
||||
}
|
||||
|
||||
result, err := r.ormRepo.FindAll(ctx, filter, pagination)
|
||||
if err != nil {
|
||||
r.logger.Error("Failed to list tenders with content_xml", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return result.Items, nil
|
||||
}
|
||||
|
||||
// Update updates an existing tender
|
||||
func (r *tenderRepository) Update(ctx context.Context, tender *Tender) error {
|
||||
tender.UpdatedAt = time.Now().Unix()
|
||||
|
||||
@@ -0,0 +1,240 @@
|
||||
package tender
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
goredis "github.com/redis/go-redis/v9"
|
||||
"tm/pkg/response"
|
||||
)
|
||||
|
||||
const (
|
||||
searchListCacheTTL = 60 * time.Second
|
||||
// searchListCacheStaleTTL bounds how long the default unfiltered list may lag after
|
||||
// writes. Only isCacheableSearchList queries use this cache (~5 min max staleness).
|
||||
searchListCacheStaleTTL = 5 * time.Minute
|
||||
searchListCacheKeyPrefix = "tender:search:list:"
|
||||
)
|
||||
|
||||
type searchListCacheEntry struct {
|
||||
expiresAt time.Time
|
||||
staleUntil time.Time
|
||||
value *SearchResponse
|
||||
}
|
||||
|
||||
type cachedSearchListPayload struct {
|
||||
Tenders []TenderResponse `json:"tenders"`
|
||||
Metadata *response.Meta `json:"metadata,omitempty"`
|
||||
}
|
||||
|
||||
func searchResponseToCachePayload(resp *SearchResponse) *cachedSearchListPayload {
|
||||
if resp == nil {
|
||||
return nil
|
||||
}
|
||||
return &cachedSearchListPayload{
|
||||
Tenders: resp.Tenders,
|
||||
Metadata: resp.Metadata,
|
||||
}
|
||||
}
|
||||
|
||||
func cachePayloadToSearchResponse(payload *cachedSearchListPayload) *SearchResponse {
|
||||
if payload == nil {
|
||||
return nil
|
||||
}
|
||||
return &SearchResponse{
|
||||
Tenders: payload.Tenders,
|
||||
Metadata: payload.Metadata,
|
||||
}
|
||||
}
|
||||
|
||||
func isCacheableSearchList(form *SearchForm, pagination *response.Pagination) bool {
|
||||
if form == nil || pagination == nil {
|
||||
return false
|
||||
}
|
||||
if form.HasListFilters() {
|
||||
return false
|
||||
}
|
||||
if form.CompanyID != nil && strings.TrimSpace(*form.CompanyID) != "" {
|
||||
return false
|
||||
}
|
||||
if form.CustomerID != nil && strings.TrimSpace(*form.CustomerID) != "" {
|
||||
return false
|
||||
}
|
||||
if len(form.CompanyIDs) > 0 {
|
||||
return false
|
||||
}
|
||||
if pagination.Cursor != "" || pagination.IncludeTotal || pagination.Offset != 0 {
|
||||
return false
|
||||
}
|
||||
if pagination.Limit <= 0 || pagination.Limit > 20 {
|
||||
return false
|
||||
}
|
||||
|
||||
sortBy := pagination.SortBy
|
||||
if sortBy == "" {
|
||||
sortBy = "created_at"
|
||||
}
|
||||
sortOrder := pagination.SortOrder
|
||||
if sortOrder == "" {
|
||||
sortOrder = "desc"
|
||||
}
|
||||
|
||||
return sortBy == "created_at" && sortOrder == "desc"
|
||||
}
|
||||
|
||||
func searchListCacheKey(pagination *response.Pagination, language string) string {
|
||||
return fmt.Sprintf("%d:%s", pagination.Limit, strings.ToLower(strings.TrimSpace(language)))
|
||||
}
|
||||
|
||||
func (s *tenderService) cachedSearchList(ctx context.Context, form *SearchForm, pagination *response.Pagination, language string, load func(context.Context) (*SearchResponse, error)) (*SearchResponse, error) {
|
||||
if !isCacheableSearchList(form, pagination) {
|
||||
return load(ctx)
|
||||
}
|
||||
|
||||
// Cached *SearchResponse values (including the Tenders slice) are shared across
|
||||
// concurrent callers via singleflight and stale hits; treat as read-only.
|
||||
cacheKey := searchListCacheKey(pagination, language)
|
||||
redisKey := searchListCacheKeyPrefix + cacheKey
|
||||
|
||||
if cached, ok := s.getFreshSearchListCache(cacheKey); ok {
|
||||
return cached, nil
|
||||
}
|
||||
if stale, ok := s.getStaleSearchListCache(cacheKey); ok {
|
||||
go s.refreshSearchListCache(cacheKey, redisKey, load)
|
||||
return stale, nil
|
||||
}
|
||||
if cached, ok := s.getRedisSearchListCache(ctx, redisKey); ok {
|
||||
s.storeSearchListCache(cacheKey, cached)
|
||||
go s.refreshSearchListCache(cacheKey, redisKey, load)
|
||||
return cached, nil
|
||||
}
|
||||
|
||||
out, err, _ := s.searchListCacheGroup.Do(cacheKey, func() (interface{}, error) {
|
||||
if cached, ok := s.getFreshSearchListCache(cacheKey); ok {
|
||||
return cached, nil
|
||||
}
|
||||
|
||||
result, err := load(context.WithoutCancel(ctx))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
s.storeSearchListCache(cacheKey, result)
|
||||
s.storeRedisSearchListCache(context.Background(), redisKey, result)
|
||||
return result, nil
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return out.(*SearchResponse), nil
|
||||
}
|
||||
|
||||
func (s *tenderService) refreshSearchListCache(cacheKey, redisKey string, load func(context.Context) (*SearchResponse, error)) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
|
||||
defer cancel()
|
||||
|
||||
result, err := load(ctx)
|
||||
if err != nil {
|
||||
s.logger.Warn("Failed to refresh tender search list cache", map[string]interface{}{
|
||||
"cache_key": cacheKey,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
s.storeSearchListCache(cacheKey, result)
|
||||
s.storeRedisSearchListCache(context.Background(), redisKey, result)
|
||||
}
|
||||
|
||||
func (s *tenderService) getFreshSearchListCache(cacheKey string) (*SearchResponse, bool) {
|
||||
s.searchListCacheMu.Lock()
|
||||
defer s.searchListCacheMu.Unlock()
|
||||
|
||||
entry, ok := s.searchListCache[cacheKey]
|
||||
if !ok || time.Now().After(entry.expiresAt) {
|
||||
return nil, false
|
||||
}
|
||||
return entry.value, true
|
||||
}
|
||||
|
||||
func (s *tenderService) getStaleSearchListCache(cacheKey string) (*SearchResponse, bool) {
|
||||
s.searchListCacheMu.Lock()
|
||||
defer s.searchListCacheMu.Unlock()
|
||||
|
||||
entry, ok := s.searchListCache[cacheKey]
|
||||
if !ok || time.Now().After(entry.staleUntil) {
|
||||
if ok {
|
||||
delete(s.searchListCache, cacheKey)
|
||||
}
|
||||
return nil, false
|
||||
}
|
||||
return entry.value, true
|
||||
}
|
||||
|
||||
func (s *tenderService) storeSearchListCache(cacheKey string, value *SearchResponse) {
|
||||
now := time.Now()
|
||||
|
||||
s.searchListCacheMu.Lock()
|
||||
defer s.searchListCacheMu.Unlock()
|
||||
|
||||
s.searchListCache[cacheKey] = searchListCacheEntry{
|
||||
expiresAt: now.Add(searchListCacheTTL),
|
||||
staleUntil: now.Add(searchListCacheTTL + searchListCacheStaleTTL),
|
||||
value: value,
|
||||
}
|
||||
}
|
||||
|
||||
func (s *tenderService) getRedisSearchListCache(ctx context.Context, redisKey string) (*SearchResponse, bool) {
|
||||
if s.redisClient == nil {
|
||||
return nil, false
|
||||
}
|
||||
|
||||
raw, err := s.redisClient.Get(ctx, redisKey)
|
||||
if err != nil {
|
||||
if err != goredis.Nil {
|
||||
s.logger.Warn("Failed to read tender search list cache from Redis", map[string]interface{}{
|
||||
"cache_key": redisKey,
|
||||
"error": err.Error(),
|
||||
})
|
||||
}
|
||||
return nil, false
|
||||
}
|
||||
|
||||
var cached cachedSearchListPayload
|
||||
if err := json.Unmarshal([]byte(raw), &cached); err != nil {
|
||||
s.logger.Warn("Failed to decode tender search list cache from Redis", map[string]interface{}{
|
||||
"cache_key": redisKey,
|
||||
"error": err.Error(),
|
||||
})
|
||||
_ = s.redisClient.Del(ctx, redisKey)
|
||||
return nil, false
|
||||
}
|
||||
|
||||
return cachePayloadToSearchResponse(&cached), true
|
||||
}
|
||||
|
||||
func (s *tenderService) storeRedisSearchListCache(ctx context.Context, redisKey string, value *SearchResponse) {
|
||||
if s.redisClient == nil || value == nil {
|
||||
return
|
||||
}
|
||||
|
||||
encoded, err := json.Marshal(searchResponseToCachePayload(value))
|
||||
if err != nil {
|
||||
s.logger.Warn("Failed to encode tender search list cache for Redis", map[string]interface{}{
|
||||
"cache_key": redisKey,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
if err := s.redisClient.Set(ctx, redisKey, string(encoded), searchListCacheTTL+searchListCacheStaleTTL); err != nil {
|
||||
s.logger.Warn("Failed to store tender search list cache in Redis", map[string]interface{}{
|
||||
"cache_key": redisKey,
|
||||
"error": err.Error(),
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"path"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
"tm/internal/company"
|
||||
"tm/internal/customer"
|
||||
@@ -20,6 +21,7 @@ import (
|
||||
"tm/pkg/response"
|
||||
|
||||
"golang.org/x/sync/errgroup"
|
||||
"golang.org/x/sync/singleflight"
|
||||
)
|
||||
|
||||
// AISummarizerClient defines the interface for on-demand AI operations.
|
||||
@@ -95,6 +97,9 @@ type tenderService struct {
|
||||
redisClient redis.Client
|
||||
pageCacheTTL time.Duration
|
||||
defaultLanguage string
|
||||
searchListCacheMu sync.Mutex
|
||||
searchListCache map[string]searchListCacheEntry
|
||||
searchListCacheGroup singleflight.Group
|
||||
}
|
||||
|
||||
const (
|
||||
@@ -136,6 +141,7 @@ func NewService(
|
||||
redisClient: redisClient,
|
||||
pageCacheTTL: recommendationPageCacheTTL,
|
||||
defaultLanguage: strings.ToLower(defaultLanguage),
|
||||
searchListCache: make(map[string]searchListCacheEntry),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -922,8 +928,13 @@ func (s *tenderService) Search(ctx context.Context, form *SearchForm, pagination
|
||||
"to": form.DeadlineTo,
|
||||
})
|
||||
|
||||
// Profile keywords are not used for tender search.
|
||||
language := s.pickResponseLanguage(form.Language)
|
||||
return s.cachedSearchList(ctx, form, pagination, language, func(loadCtx context.Context) (*SearchResponse, error) {
|
||||
return s.searchUncached(loadCtx, form, pagination, language)
|
||||
})
|
||||
}
|
||||
|
||||
func (s *tenderService) searchUncached(ctx context.Context, form *SearchForm, pagination *response.Pagination, language string) (*SearchResponse, error) {
|
||||
page, err := s.repository.Search(ctx, form, pagination)
|
||||
if err != nil {
|
||||
s.logger.Error("Failed to list tenders", map[string]interface{}{
|
||||
@@ -936,7 +947,7 @@ func (s *tenderService) Search(ctx context.Context, form *SearchForm, pagination
|
||||
if form.CompanyID == nil || *form.CompanyID == "" {
|
||||
tenderResponses := make([]TenderResponse, 0, len(page.Items))
|
||||
for _, tender := range page.Items {
|
||||
tenderResponses = append(tenderResponses, *tender.ToResponseWithLanguage(s.pickResponseLanguage(form.Language)))
|
||||
tenderResponses = append(tenderResponses, *tender.ToResponseWithLanguage(language))
|
||||
}
|
||||
|
||||
return &SearchResponse{
|
||||
@@ -955,7 +966,7 @@ func (s *tenderService) Search(ctx context.Context, form *SearchForm, pagination
|
||||
// Continue without match percentage if company not found
|
||||
tenderResponses := make([]TenderResponse, 0, len(page.Items))
|
||||
for _, tender := range page.Items {
|
||||
tenderResponses = append(tenderResponses, *tender.ToResponseWithLanguage(s.pickResponseLanguage(form.Language)))
|
||||
tenderResponses = append(tenderResponses, *tender.ToResponseWithLanguage(language))
|
||||
}
|
||||
|
||||
return &SearchResponse{
|
||||
@@ -978,7 +989,7 @@ func (s *tenderService) Search(ctx context.Context, form *SearchForm, pagination
|
||||
|
||||
tenderResponses := make([]TenderResponse, 0, len(page.Items))
|
||||
for _, tender := range page.Items {
|
||||
tenderResponses = append(tenderResponses, *tender.ToResponseWithLanguage(s.pickResponseLanguage(form.Language)))
|
||||
tenderResponses = append(tenderResponses, *tender.ToResponseWithLanguage(language))
|
||||
}
|
||||
|
||||
return &SearchResponse{
|
||||
|
||||
Reference in New Issue
Block a user