Mazyar
8dbd9927b0
continuous-integration/drone/push Build is passing
- Updated the dashboard service to integrate Redis caching for improved performance in statistics retrieval. - Modified the NewService function to accept a Redis client, enabling caching of dashboard statistics. - Implemented logic to retrieve statistics from Redis, falling back to the database if necessary, and introduced a background process to warm the cache. - Enhanced error handling and logging for Redis operations to ensure robust statistics management. - Increased cache duration for scraped documents and adjusted timeout settings for MongoDB queries to optimize performance. This update significantly improves the responsiveness and efficiency of the dashboard by leveraging Redis for caching statistics.
502 lines
13 KiB
Go
502 lines
13 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 = 60 * time.Second
|
|
statisticsCacheTTL = 5 * time.Minute
|
|
statisticsStaleGraceTTL = 15 * time.Minute
|
|
)
|
|
|
|
// 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[int64]cacheEntry[*SummaryResponse]
|
|
summaryGroup singleflight.Group
|
|
statisticsMu sync.Mutex
|
|
statistics map[int]cacheEntry[*StatisticsReportResponse]
|
|
statisticsGroup singleflight.Group
|
|
}
|
|
|
|
// 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[int64]cacheEntry[*SummaryResponse]),
|
|
statistics: make(map[int]cacheEntry[*StatisticsReportResponse]),
|
|
}
|
|
go s.warmStatisticsCache()
|
|
return s
|
|
}
|
|
|
|
func (s *service) Summary(ctx context.Context, query SummaryQuery) (*SummaryResponse, error) {
|
|
windowHours := closingWindowHours(query.ClosingWindow)
|
|
windowSec := int64(windowHours) * 3600
|
|
|
|
if cached, ok := s.cachedSummary(windowSec); ok {
|
|
return cached, nil
|
|
}
|
|
|
|
cacheKey := strconv.FormatInt(windowSec, 10)
|
|
out, err, _ := s.summaryGroup.Do(cacheKey, func() (interface{}, error) {
|
|
if cached, ok := s.cachedSummary(windowSec); ok {
|
|
return cached, nil
|
|
}
|
|
|
|
s.logger.Info("Fetching dashboard summary", map[string]interface{}{
|
|
"closing_window_hours": windowHours,
|
|
})
|
|
|
|
result, err := s.repo.Summary(context.WithoutCancel(ctx), windowSec)
|
|
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(windowSec, 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)
|
|
startUnix := trendStartUnix(days)
|
|
|
|
s.logger.Info("Fetching dashboard trend", map[string]interface{}{
|
|
"days": days,
|
|
"metric": metric,
|
|
})
|
|
|
|
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(),
|
|
})
|
|
return nil, fmt.Errorf("dashboard trend: %w", err)
|
|
}
|
|
|
|
series := fillTrendSeries(days, counts)
|
|
|
|
return &TrendResponse{
|
|
Metric: metric,
|
|
Days: days,
|
|
Series: series,
|
|
}, nil
|
|
}
|
|
|
|
func (s *service) Countries(ctx context.Context, query CountriesQuery) (*CountriesResponse, error) {
|
|
limit := countriesLimit(query.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 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)
|
|
|
|
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)
|
|
}
|
|
|
|
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
|
|
|
|
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 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(),
|
|
})
|
|
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
|
|
}
|
|
|
|
result, err := s.loadStatistics(ctx, days)
|
|
if err == nil {
|
|
return result, nil
|
|
}
|
|
|
|
if redisCached, ok := s.getRedisStatistics(ctx, days); ok {
|
|
s.storeStatistics(days, redisCached)
|
|
return redisCached, nil
|
|
}
|
|
|
|
s.logger.Error("Serving empty dashboard statistics after fetch failure", map[string]interface{}{
|
|
"days": days,
|
|
"error": err.Error(),
|
|
})
|
|
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{
|
|
ScrapedTED: fillTrendSeries(days, nil),
|
|
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) cachedSummary(windowSec int64) (*SummaryResponse, bool) {
|
|
now := time.Now()
|
|
|
|
s.summaryMu.Lock()
|
|
defer s.summaryMu.Unlock()
|
|
|
|
entry, ok := s.summary[windowSec]
|
|
if !ok || now.After(entry.expiresAt) {
|
|
if ok {
|
|
delete(s.summary, windowSec)
|
|
}
|
|
return nil, false
|
|
}
|
|
|
|
return entry.value, true
|
|
}
|
|
|
|
func (s *service) storeSummary(windowSec int64, value *SummaryResponse) {
|
|
s.summaryMu.Lock()
|
|
defer s.summaryMu.Unlock()
|
|
|
|
s.summary[windowSec] = cacheEntry[*SummaryResponse]{
|
|
expiresAt: time.Now().Add(summaryCacheTTL),
|
|
staleUntil: time.Now().Add(summaryCacheTTL),
|
|
value: value,
|
|
}
|
|
}
|
|
|
|
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
|
|
}
|