diff --git a/internal/dashboard/repository.go b/internal/dashboard/repository.go index 6d17aff..4d11691 100644 --- a/internal/dashboard/repository.go +++ b/internal/dashboard/repository.go @@ -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( diff --git a/internal/dashboard/repository_test.go b/internal/dashboard/repository_test.go index 222c1a1..36f2eeb 100644 --- a/internal/dashboard/repository_test.go +++ b/internal/dashboard/repository_test.go @@ -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) + } +} diff --git a/internal/dashboard/service.go b/internal/dashboard/service.go index b92e1a3..7bffc25 100644 --- a/internal/dashboard/service.go +++ b/internal/dashboard/service.go @@ -60,19 +60,30 @@ 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. func NewService(repo Repository, log logger.Logger, redisClient redis.Client) Service { s := &service{ - repo: repo, - logger: log, - redis: redisClient, - summary: make(map[string]cacheEntry[*SummaryResponse]), - statistics: make(map[int]cacheEntry[*StatisticsReportResponse]), + repo: repo, + logger: log, + 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,102 +140,134 @@ 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) - startUnix := trendStartUnix(days) + cacheKey := fmt.Sprintf("%d:%s", days, metric) - s.logger.Info("Fetching dashboard trend", map[string]interface{}{ - "days": days, - "metric": metric, - }) + return s.trendCache.Get(ctx, cacheKey, func(loadCtx context.Context) (*TrendResponse, error) { + startUnix := trendStartUnix(days) - counts, err := s.repo.Trend(ctx, days, metric, startUnix) - if err != nil { - s.logger.Error("Failed to fetch dashboard trend", map[string]interface{}{ - "error": err.Error(), + s.logger.Info("Fetching dashboard trend", map[string]interface{}{ + "days": days, + "metric": metric, }) - return nil, fmt.Errorf("dashboard trend: %w", err) - } - series := fillTrendSeries(days, counts) + 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(), + }) + return nil, fmt.Errorf("dashboard trend: %w", err) + } - return &TrendResponse{ - Metric: metric, - Days: days, - Series: series, - }, nil + return &TrendResponse{ + Metric: metric, + Days: days, + Series: fillTrendSeries(days, counts), + }, nil + }) } func (s *service) Countries(ctx context.Context, query CountriesQuery) (*CountriesResponse, error) { limit := countriesLimit(query.Limit) + cacheKey := strconv.Itoa(limit) - s.logger.Info("Fetching dashboard countries", map[string]interface{}{ - "limit": limit, - }) - - out, err := s.repo.Countries(ctx, limit) - if err != nil { - s.logger.Error("Failed to fetch dashboard countries", map[string]interface{}{ - "error": err.Error(), + return s.countriesCache.Get(ctx, cacheKey, func(loadCtx context.Context) (*CountriesResponse, error) { + s.logger.Info("Fetching dashboard countries", map[string]interface{}{ + "limit": limit, }) - return nil, fmt.Errorf("dashboard countries: %w", err) - } - return out, nil + out, err := s.repo.Countries(loadCtx, limit) + if err != nil { + s.logger.Error("Failed to fetch dashboard countries", map[string]interface{}{ + "error": err.Error(), + }) + return nil, fmt.Errorf("dashboard countries: %w", err) + } + + return out, nil + }) } func (s *service) NoticeTypes(ctx context.Context) (*NoticeTypesResponse, error) { - s.logger.Info("Fetching dashboard notice types", nil) + 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) - if err != nil { - s.logger.Error("Failed to fetch dashboard notice types", map[string]interface{}{ - "error": err.Error(), - }) - return nil, fmt.Errorf("dashboard notice types: %w", err) - } + out, err := s.repo.NoticeTypes(loadCtx) + if err != nil { + s.logger.Error("Failed to fetch dashboard notice types", map[string]interface{}{ + "error": err.Error(), + }) + return nil, fmt.Errorf("dashboard notice types: %w", err) + } - return out, nil + 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) - s.logger.Info("Fetching dashboard closing soon", map[string]interface{}{ - "limit": limit, - "window": windowHours, - }) - - items, err := s.repo.ClosingSoon(ctx, limit, windowSec) - if err != nil { - s.logger.Error("Failed to fetch dashboard closing soon", map[string]interface{}{ - "error": err.Error(), + 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, }) - return nil, fmt.Errorf("dashboard closing soon: %w", err) - } - return &ClosingSoonResponse{Items: items}, nil + 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(), + }) + return nil, fmt.Errorf("dashboard closing soon: %w", err) + } + + return &ClosingSoonResponse{Items: items}, nil + }) } func (s *service) Recent(ctx context.Context, query RecentQuery) (*RecentResponse, error) { limit := listLimit(query.Limit) - - s.logger.Info("Fetching dashboard recent tenders", map[string]interface{}{ - "limit": limit, - }) - - items, nextCursor, err := s.repo.Recent(ctx, limit, strings.TrimSpace(query.Cursor)) - if err != nil { - s.logger.Error("Failed to fetch dashboard recent tenders", map[string]interface{}{ - "error": err.Error(), + cursor := strings.TrimSpace(query.Cursor) + if cursor != "" { + s.logger.Info("Fetching dashboard recent tenders", map[string]interface{}{ + "limit": limit, }) - return nil, fmt.Errorf("dashboard recent: %w", err) + + 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(), + }) + return nil, fmt.Errorf("dashboard recent: %w", err) + } + + return &RecentResponse{ + Items: items, + NextCursor: nextCursor, + }, nil } - return &RecentResponse{ - 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() diff --git a/internal/dashboard/widget_cache.go b/internal/dashboard/widget_cache.go new file mode 100644 index 0000000..e0b0b02 --- /dev/null +++ b/internal/dashboard/widget_cache.go @@ -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(), + }) + } +} diff --git a/internal/notification/entity.go b/internal/notification/entity.go index 95292b8..7782444 100644 --- a/internal/notification/entity.go +++ b/internal/notification/entity.go @@ -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 diff --git a/internal/notification/service.go b/internal/notification/service.go index e14e072..f2f1fd8 100644 --- a/internal/notification/service.go +++ b/internal/notification/service.go @@ -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 } - if slices.Contains(req.Channels, DeliveryChannelEmail) && recipient.Email != "" { + + methods := map[string]string{} + switch channel { + case DeliveryChannelEmail: + if recipient.Email == "" { + return nil + } methods["email"] = recipient.Email - } - if slices.Contains(req.Channels, DeliveryChannelPush) && len(recipient.DeviceTokens) > 0 { + case DeliveryChannelPush: + if len(recipient.DeviceTokens) == 0 { + return nil + } 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{ diff --git a/internal/tender/form.go b/internal/tender/form.go index 5055050..5eea71a 100644 --- a/internal/tender/form.go +++ b/internal/tender/form.go @@ -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"` diff --git a/internal/tender/repository.go b/internal/tender/repository.go index 5df01fb..fa0712d 100644 --- a/internal/tender/repository.go +++ b/internal/tender/repository.go @@ -129,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). diff --git a/internal/tender/search_list_cache.go b/internal/tender/search_list_cache.go new file mode 100644 index 0000000..05228e5 --- /dev/null +++ b/internal/tender/search_list_cache.go @@ -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(), + }) + } +} diff --git a/internal/tender/service.go b/internal/tender/service.go index d97dbdf..4fc3803 100644 --- a/internal/tender/service.go +++ b/internal/tender/service.go @@ -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{