Files
Mazyar c5e62a4061 Implement deadline matching logic for dashboard repository and enhance unit tests
- Added the `deadlineWithinWindowMatch` function to match raw deadline fields stored as Unix seconds or milliseconds, ensuring accurate filtering in MongoDB queries.
- Updated the `ClosingSoon` method in the repository to utilize the new deadline matching logic, improving query accuracy for upcoming deadlines.
- Introduced a new unit test, `TestDeadlineWithinWindowMatchCoversSecondsAndMilliseconds`, to validate the functionality of the deadline matching logic, ensuring both seconds and milliseconds are correctly handled.
- Enhanced comments in the `widget_cache` and `search_list_cache` files for better clarity on cache behavior and staleness.

This update improves the accuracy of deadline handling in the dashboard service and strengthens the test coverage for related functionalities.
2026-07-11 21:44:22 +03:30

785 lines
22 KiB
Go

package dashboard
import (
"context"
"encoding/json"
"fmt"
"strconv"
"strings"
"sync"
"time"
"tm/pkg/logger"
"tm/pkg/redis"
goredis "github.com/redis/go-redis/v9"
"golang.org/x/sync/singleflight"
)
const (
defaultClosingWindowHours = 168
defaultTrendDays = 14
maxTrendDays = 90
defaultCountriesLimit = 6
defaultListLimit = 5
maxListLimit = 20
summaryCacheTTL = 5 * time.Minute
summaryStaleGraceTTL = 15 * time.Minute
statisticsCacheTTL = 5 * time.Minute
statisticsStaleGraceTTL = 15 * time.Minute
// statisticsColdLoadWait bounds how long a cold-cache request waits for a fresh
// load before falling back to a placeholder. Mongo-only loads (MinIO scope
// already cached) comfortably finish within this window; a first-ever MinIO
// bucket scan does not, so we don't block the request for that.
statisticsColdLoadWait = 4 * time.Second
)
// Service defines dashboard business operations.
type Service interface {
Summary(ctx context.Context, query SummaryQuery) (*SummaryResponse, error)
Trend(ctx context.Context, query TrendQuery) (*TrendResponse, error)
Countries(ctx context.Context, query CountriesQuery) (*CountriesResponse, error)
NoticeTypes(ctx context.Context) (*NoticeTypesResponse, error)
ClosingSoon(ctx context.Context, query ClosingSoonQuery) (*ClosingSoonResponse, error)
Recent(ctx context.Context, query RecentQuery) (*RecentResponse, error)
Statistics(ctx context.Context, query StatisticsQuery) (*StatisticsReportResponse, error)
}
type cacheEntry[T any] struct {
expiresAt time.Time
staleUntil time.Time
value T
}
type service struct {
repo Repository
logger logger.Logger
redis redis.Client
summaryMu sync.Mutex
summary map[string]cacheEntry[*SummaryResponse]
summaryGroup singleflight.Group
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]),
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
}
func (s *service) Summary(ctx context.Context, query SummaryQuery) (*SummaryResponse, error) {
windowHours := closingWindowHours(query.ClosingWindow)
windowSec := int64(windowHours) * 3600
days := trendDays(query.Days)
cacheKey := summaryCacheKey(windowSec, days)
if cached, ok := s.cachedSummary(cacheKey); ok {
return cached, nil
}
if stale, ok := s.staleSummary(cacheKey); ok {
go s.refreshSummary(windowSec, days)
return stale, nil
}
if redisCached, ok := s.getRedisSummary(ctx, cacheKey); ok {
s.storeSummary(cacheKey, redisCached)
go s.refreshSummary(windowSec, days)
return redisCached, nil
}
out, err, _ := s.summaryGroup.Do(cacheKey, func() (interface{}, error) {
if cached, ok := s.cachedSummary(cacheKey); ok {
return cached, nil
}
s.logger.Info("Fetching dashboard summary", map[string]interface{}{
"closing_window_hours": windowHours,
"days": days,
})
result, err := s.repo.Summary(context.WithoutCancel(ctx), windowSec, days)
if err != nil {
s.logger.Error("Failed to fetch dashboard summary", map[string]interface{}{
"error": err.Error(),
})
return nil, fmt.Errorf("dashboard summary: %w", err)
}
s.storeSummary(cacheKey, result)
s.storeRedisSummary(context.Background(), cacheKey, result)
return result, nil
})
if err != nil {
return nil, err
}
return out.(*SummaryResponse), nil
}
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{}{
"days": days,
"metric": metric,
})
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: 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(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) {
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(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
})
}
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(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)
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, 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
}
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) {
days := trendDays(query.Days)
if cached, ok := s.cachedStatistics(days); ok {
return cached, nil
}
if stale, ok := s.staleStatistics(days); ok {
go s.refreshStatistics(days)
return stale, nil
}
if redisCached, ok := s.getRedisStatistics(ctx, days); ok {
s.storeStatistics(days, redisCached)
go s.refreshStatistics(days)
return redisCached, nil
}
// Cold cache: the underlying load performs a full MinIO bucket scan plus Mongo
// aggregations that can take minutes on a first run. Start the load in the
// background (deduped via singleflight, and detached from this request's
// context so it keeps running even if we give up waiting) and wait briefly for
// it. If it finishes in time we return real data; otherwise we serve a
// placeholder so the endpoint stays fast, and the load keeps warming the cache
// for the next request.
resultCh := make(chan *StatisticsReportResponse, 1)
go func() {
result, err := s.loadStatistics(context.Background(), days)
if err != nil {
resultCh <- nil
return
}
resultCh <- result
}()
select {
case result := <-resultCh:
if result != nil {
return result, nil
}
case <-time.After(statisticsColdLoadWait):
s.logger.Info("Serving placeholder dashboard statistics while cache warms", map[string]interface{}{
"days": days,
})
}
return emptyStatisticsReport(days), nil
}
func (s *service) warmStatisticsCache() {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
if cached, ok := s.getRedisStatistics(ctx, defaultTrendDays); ok {
s.storeStatistics(defaultTrendDays, cached)
s.logger.Info("Dashboard statistics cache warmed from Redis", map[string]interface{}{
"days": defaultTrendDays,
})
return
}
go s.refreshStatistics(defaultTrendDays)
}
func emptyStatisticsReport(days int) *StatisticsReportResponse {
return &StatisticsReportResponse{
Days: days,
GeneratedAt: time.Now().UTC().Unix(),
Daily: StatisticsDailySeries{
ScrapedDocuments: fillTrendSeries(days, nil),
TranslatedNotices: fillTrendSeries(days, nil),
},
}
}
func (s *service) refreshStatistics(days int) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
defer cancel()
_, _ = s.loadStatistics(ctx, days)
}
func (s *service) loadStatistics(ctx context.Context, days int) (*StatisticsReportResponse, error) {
cacheKey := strconv.Itoa(days)
out, err, _ := s.statisticsGroup.Do(cacheKey, func() (interface{}, error) {
if cached, ok := s.cachedStatistics(days); ok {
return cached, nil
}
startUnix := trendStartUnix(days)
s.logger.Info("Fetching dashboard statistics report", map[string]interface{}{
"days": days,
})
result, err := s.repo.Statistics(context.WithoutCancel(ctx), days, startUnix)
if err != nil {
s.logger.Error("Failed to fetch dashboard statistics report", map[string]interface{}{
"error": err.Error(),
})
return nil, fmt.Errorf("dashboard statistics: %w", err)
}
s.storeStatistics(days, result)
s.storeRedisStatistics(context.Background(), days, result)
return result, nil
})
if err != nil {
return nil, err
}
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()
windowSec := int64(defaultClosingWindowHours) * 3600
cacheKey := summaryCacheKey(windowSec, defaultTrendDays)
if cached, ok := s.getRedisSummary(ctx, cacheKey); ok {
s.storeSummary(cacheKey, cached)
s.logger.Info("Dashboard summary cache warmed from Redis", map[string]interface{}{
"closing_window_hours": defaultClosingWindowHours,
"days": defaultTrendDays,
})
return
}
go s.refreshSummary(windowSec, defaultTrendDays)
}
func (s *service) refreshSummary(windowSec int64, days int) {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer cancel()
_, _ = s.loadSummary(ctx, windowSec, days)
}
func (s *service) loadSummary(ctx context.Context, windowSec int64, days int) (*SummaryResponse, error) {
cacheKey := summaryCacheKey(windowSec, days)
out, err, _ := s.summaryGroup.Do(cacheKey, func() (interface{}, error) {
if cached, ok := s.cachedSummary(cacheKey); ok {
return cached, nil
}
s.logger.Info("Refreshing dashboard summary cache", map[string]interface{}{
"closing_window_sec": windowSec,
"days": days,
})
result, err := s.repo.Summary(context.WithoutCancel(ctx), windowSec, days)
if err != nil {
s.logger.Error("Failed to refresh dashboard summary", map[string]interface{}{
"error": err.Error(),
})
return nil, fmt.Errorf("dashboard summary: %w", err)
}
s.storeSummary(cacheKey, result)
s.storeRedisSummary(context.Background(), cacheKey, result)
return result, nil
})
if err != nil {
return nil, err
}
return out.(*SummaryResponse), nil
}
func summaryCacheKey(windowSec int64, days int) string {
return fmt.Sprintf("%d:%d", windowSec, days)
}
func summaryRedisKey(cacheKey string) string {
return fmt.Sprintf("dashboard:summary:%s", cacheKey)
}
func (s *service) cachedSummary(cacheKey string) (*SummaryResponse, bool) {
now := time.Now()
s.summaryMu.Lock()
defer s.summaryMu.Unlock()
entry, ok := s.summary[cacheKey]
if !ok || now.After(entry.expiresAt) {
return nil, false
}
return entry.value, true
}
func (s *service) staleSummary(cacheKey string) (*SummaryResponse, bool) {
now := time.Now()
s.summaryMu.Lock()
defer s.summaryMu.Unlock()
entry, ok := s.summary[cacheKey]
if !ok || now.After(entry.staleUntil) {
if ok {
delete(s.summary, cacheKey)
}
return nil, false
}
return entry.value, true
}
func (s *service) storeSummary(cacheKey string, value *SummaryResponse) {
now := time.Now()
s.summaryMu.Lock()
defer s.summaryMu.Unlock()
s.summary[cacheKey] = cacheEntry[*SummaryResponse]{
expiresAt: now.Add(summaryCacheTTL),
staleUntil: now.Add(summaryCacheTTL + summaryStaleGraceTTL),
value: value,
}
}
func (s *service) getRedisSummary(ctx context.Context, cacheKey string) (*SummaryResponse, bool) {
if s.redis == nil {
return nil, false
}
raw, err := s.redis.Get(ctx, summaryRedisKey(cacheKey))
if err != nil {
if err != goredis.Nil {
s.logger.Warn("Failed to read dashboard summary cache from Redis", map[string]interface{}{
"cache_key": cacheKey,
"error": err.Error(),
})
}
return nil, false
}
var report SummaryResponse
if err := json.Unmarshal([]byte(raw), &report); err != nil {
s.logger.Warn("Failed to decode dashboard summary cache from Redis", map[string]interface{}{
"cache_key": cacheKey,
"error": err.Error(),
})
_ = s.redis.Del(ctx, summaryRedisKey(cacheKey))
return nil, false
}
return &report, true
}
func (s *service) storeRedisSummary(ctx context.Context, cacheKey string, value *SummaryResponse) {
if s.redis == nil || value == nil {
return
}
encoded, err := json.Marshal(value)
if err != nil {
s.logger.Warn("Failed to encode dashboard summary for Redis", map[string]interface{}{
"cache_key": cacheKey,
"error": err.Error(),
})
return
}
ttl := summaryCacheTTL + summaryStaleGraceTTL
if err := s.redis.Set(ctx, summaryRedisKey(cacheKey), string(encoded), ttl); err != nil {
s.logger.Warn("Failed to store dashboard summary in Redis", map[string]interface{}{
"cache_key": cacheKey,
"error": err.Error(),
})
}
}
func (s *service) cachedStatistics(days int) (*StatisticsReportResponse, bool) {
now := time.Now()
s.statisticsMu.Lock()
defer s.statisticsMu.Unlock()
entry, ok := s.statistics[days]
if !ok || now.After(entry.expiresAt) {
return nil, false
}
return entry.value, true
}
func (s *service) staleStatistics(days int) (*StatisticsReportResponse, bool) {
now := time.Now()
s.statisticsMu.Lock()
defer s.statisticsMu.Unlock()
entry, ok := s.statistics[days]
if !ok || now.After(entry.staleUntil) {
if ok {
delete(s.statistics, days)
}
return nil, false
}
return entry.value, true
}
func (s *service) storeStatistics(days int, value *StatisticsReportResponse) {
now := time.Now()
s.statisticsMu.Lock()
defer s.statisticsMu.Unlock()
s.statistics[days] = cacheEntry[*StatisticsReportResponse]{
expiresAt: now.Add(statisticsCacheTTL),
staleUntil: now.Add(statisticsCacheTTL + statisticsStaleGraceTTL),
value: value,
}
}
func statisticsRedisKey(days int) string {
return fmt.Sprintf("dashboard:statistics:%d", days)
}
func (s *service) getRedisStatistics(ctx context.Context, days int) (*StatisticsReportResponse, bool) {
if s.redis == nil {
return nil, false
}
raw, err := s.redis.Get(ctx, statisticsRedisKey(days))
if err != nil {
if err != goredis.Nil {
s.logger.Warn("Failed to read dashboard statistics cache from Redis", map[string]interface{}{
"days": days,
"error": err.Error(),
})
}
return nil, false
}
var report StatisticsReportResponse
if err := json.Unmarshal([]byte(raw), &report); err != nil {
s.logger.Warn("Failed to decode dashboard statistics cache from Redis", map[string]interface{}{
"days": days,
"error": err.Error(),
})
_ = s.redis.Del(ctx, statisticsRedisKey(days))
return nil, false
}
return &report, true
}
func (s *service) storeRedisStatistics(ctx context.Context, days int, value *StatisticsReportResponse) {
if s.redis == nil || value == nil {
return
}
encoded, err := json.Marshal(value)
if err != nil {
s.logger.Warn("Failed to encode dashboard statistics for Redis", map[string]interface{}{
"days": days,
"error": err.Error(),
})
return
}
ttl := statisticsCacheTTL + statisticsStaleGraceTTL
if err := s.redis.Set(ctx, statisticsRedisKey(days), string(encoded), ttl); err != nil {
s.logger.Warn("Failed to store dashboard statistics in Redis", map[string]interface{}{
"days": days,
"error": err.Error(),
})
}
}
func closingWindowHours(hours int) int {
if hours <= 0 {
return defaultClosingWindowHours
}
return hours
}
func trendDays(days int) int {
if days <= 0 {
return defaultTrendDays
}
if days > maxTrendDays {
return maxTrendDays
}
return days
}
func countriesLimit(limit int) int {
if limit <= 0 {
return defaultCountriesLimit
}
return limit
}
func listLimit(limit int) int {
if limit <= 0 {
return defaultListLimit
}
if limit > maxListLimit {
return maxListLimit
}
return limit
}
func normalizeTrendMetric(metric string) string {
switch strings.ToLower(strings.TrimSpace(metric)) {
case "published", "awarded":
return strings.ToLower(strings.TrimSpace(metric))
default:
return "created"
}
}
func trendStartUnix(days int) int64 {
now := time.Now().UTC()
end := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC)
start := end.AddDate(0, 0, -(days - 1))
return start.Unix()
}
func fillTrendSeries(days int, counts map[string]int64) []TrendPoint {
now := time.Now().UTC()
end := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC)
start := end.AddDate(0, 0, -(days-1))
series := make([]TrendPoint, 0, days)
for d := start; !d.After(end); d = d.AddDate(0, 0, 1) {
date := d.Format("2006-01-02")
series = append(series, TrendPoint{
Date: date,
Count: counts[date],
})
}
return series
}