Enhance dashboard summary and statistics with new fields and caching improvements
continuous-integration/drone/push Build is passing

- Added new field `Days` and `ScrapedTED` to `SummaryResponse` for tracking daily TED scrape counts.
- Updated `SummaryQuery` to include `Days` parameter for querying scraped TED data.
- Modified `Summary` handler to accept and process the new `Days` parameter.
- Refactored `Summary` method in the repository to support the new `Days` parameter and improved aggregation logic.
- Enhanced caching logic in the service layer to utilize a composite cache key based on `closingWindowSec` and `days`, improving cache management and retrieval efficiency.

This update improves the dashboard's functionality by providing more detailed insights into TED scraping activities and optimizing the caching strategy for better performance.
This commit is contained in:
Mazyar
2026-07-01 23:03:23 +03:30
parent eeafe2a625
commit 541d08a014
6 changed files with 347 additions and 119 deletions
+2 -1
View File
@@ -10,6 +10,8 @@ type SummaryResponse struct {
ClosingSoon int64 `json:"closing_soon"`
TotalEstimatedValue float64 `json:"total_estimated_value"`
ValueCurrency string `json:"value_currency"`
Days int `json:"days"`
ScrapedTED []TrendPoint `json:"scraped_ted"`
GeneratedAt int64 `json:"generated_at"`
}
@@ -96,7 +98,6 @@ type StatisticsReportResponse struct {
// StatisticsDailySeries contains per-day chart series.
type StatisticsDailySeries struct {
ScrapedTED []TrendPoint `json:"scraped_ted"`
ScrapedDocuments []TrendPoint `json:"scraped_documents"`
TranslatedNotices []TrendPoint `json:"translated_notices"`
}
+1
View File
@@ -3,6 +3,7 @@ package dashboard
// SummaryQuery binds query params for GET /dashboard/summary.
type SummaryQuery struct {
ClosingWindow int `query:"closing_window"`
Days int `query:"days"`
}
// TrendQuery binds query params for GET /dashboard/trend.
+5 -3
View File
@@ -19,10 +19,11 @@ func NewHandler(service Service) *Handler {
// Summary returns top-level dashboard counters.
// @Summary Dashboard summary
// @Description Top-level counters for hero pills and stat cards
// @Description Top-level counters for hero pills and stat cards, including daily TED scrape counts
// @Tags Admin-Dashboard
// @Produce json
// @Param closing_window query int false "Hours considered closing soon (default 168)"
// @Param days query int false "Days back for scraped TED daily series (default 14, max 90)"
// @Success 200 {object} response.APIResponse{data=SummaryResponse}
// @Failure 500 {object} response.APIResponse
// @Router /admin/v1/dashboard/summary [get]
@@ -30,6 +31,7 @@ func NewHandler(service Service) *Handler {
func (h *Handler) Summary(c echo.Context) error {
query := SummaryQuery{
ClosingWindow: parseIntQuery(c, "closing_window", 0),
Days: parseIntQuery(c, "days", 0),
}
out, err := h.service.Summary(c.Request().Context(), query)
@@ -37,7 +39,7 @@ func (h *Handler) Summary(c echo.Context) error {
return response.InternalServerError(c, "Failed to load dashboard summary")
}
setPrivateCache(c, 60)
setPrivateCache(c, 300)
return response.Success(c, out, "Dashboard summary retrieved successfully")
}
@@ -163,7 +165,7 @@ func (h *Handler) Recent(c echo.Context) error {
// Statistics returns scraping and translation statistics for charts.
// @Summary Dashboard statistics report
// @Description Daily and lifetime statistics for TED scraping, document scraping, and AI translation
// @Description Daily and lifetime statistics for document scraping and AI translation
// @Tags Admin-Dashboard
// @Produce json
// @Param days query int false "Days back for daily series (default 14, max 90)"
+164 -67
View File
@@ -31,7 +31,7 @@ const defaultValueCurrency = "EUR"
// Repository defines dashboard data access.
type Repository interface {
Summary(ctx context.Context, closingWindowSec int64) (*SummaryResponse, error)
Summary(ctx context.Context, closingWindowSec int64, days int) (*SummaryResponse, error)
Trend(ctx context.Context, days int, metric string, startUnix int64) (map[string]int64, error)
Countries(ctx context.Context, limit int) (*CountriesResponse, error)
NoticeTypes(ctx context.Context) (*NoticeTypesResponse, error)
@@ -68,89 +68,186 @@ func tenderCollectionName() string {
return "tenders"
}
func (r *repository) Summary(ctx context.Context, closingWindowSec int64) (*SummaryResponse, error) {
func (r *repository) Summary(ctx context.Context, closingWindowSec int64, days int) (*SummaryResponse, error) {
now := time.Now().Unix()
windowEnd := now + closingWindowSec
nowUTC := time.Now().UTC()
endDay := time.Date(nowUTC.Year(), nowUTC.Month(), nowUTC.Day(), 0, 0, 0, 0, time.UTC)
startDay := endDay.AddDate(0, 0, -(days - 1))
var (
statusCounts summaryStatusCounts
valueCurrency string
totalValue float64
scrapedTED map[string]int64
wg sync.WaitGroup
mu sync.Mutex
firstErr error
)
recordErr := func(section string, err error) {
mu.Lock()
defer mu.Unlock()
if firstErr == nil {
firstErr = fmt.Errorf("%s: %w", section, err)
}
}
wg.Add(3)
go func() {
defer wg.Done()
counts, err := r.summaryStatusCounts(ctx, now, windowEnd)
if err != nil {
recordErr("status_counts", err)
return
}
statusCounts = counts
}()
go func() {
defer wg.Done()
currency, value, err := r.summaryCurrencyValue(ctx)
if err != nil {
recordErr("currency_value", err)
return
}
valueCurrency, totalValue = currency, value
}()
go func() {
defer wg.Done()
qctx, cancel := context.WithTimeout(ctx, statisticsMongoQueryTimeout)
defer cancel()
counts, err := r.metricsCounter.GetTEDNoticeScrapedDailyCounts(qctx, startDay, endDay)
if err != nil {
if r.logger != nil {
r.logger.Warn("Dashboard summary scraped TED daily counts failed", map[string]interface{}{
"error": err.Error(),
})
}
scrapedTED = map[string]int64{}
return
}
scrapedTED = counts
}()
wg.Wait()
if firstErr != nil {
return nil, fmt.Errorf("dashboard summary aggregation: %w", firstErr)
}
return &SummaryResponse{
TotalTenders: statusCounts.total,
ActiveTenders: statusCounts.active,
AwardedTenders: statusCounts.awarded,
ExpiredTenders: statusCounts.expired,
CancelledTenders: statusCounts.cancelled,
ClosingSoon: statusCounts.closingSoon,
TotalEstimatedValue: totalValue,
ValueCurrency: valueCurrency,
Days: days,
ScrapedTED: fillTrendSeries(days, scrapedTED),
GeneratedAt: now,
}, nil
}
type summaryStatusCounts struct {
total int64
active int64
awarded int64
expired int64
cancelled int64
closingSoon int64
}
func (r *repository) summaryStatusCounts(ctx context.Context, now, windowEnd int64) (summaryStatusCounts, error) {
pipeline := mongo.Pipeline{
{{Key: "$facet", Value: bson.M{
"total": bson.A{
bson.M{"$count": "count"},
},
"active": bson.A{
bson.M{"$match": bson.M{"status": tender.TenderStatusActive}},
bson.M{"$count": "count"},
},
"awarded": bson.A{
bson.M{"$match": bson.M{"status": tender.TenderStatusAwarded}},
bson.M{"$count": "count"},
},
"cancelled": bson.A{
bson.M{"$match": bson.M{"status": tender.TenderStatusCancelled}},
bson.M{"$count": "count"},
},
"expired": bson.A{
bson.M{"$addFields": bson.M{"expiry_deadline": expiryDeadlineExpr()}},
bson.M{"$match": bson.M{
"status": bson.M{"$nin": []tender.TenderStatus{
tender.TenderStatusAwarded,
tender.TenderStatusCancelled,
{{Key: "$addFields", Value: bson.M{
"expiry_deadline": expiryDeadlineExpr(),
"effective_deadline": effectiveDeadlineExpr(),
}}},
{{Key: "$group", Value: bson.M{
"_id": nil,
"total": bson.M{"$sum": 1},
"active": bson.M{"$sum": bson.M{"$cond": bson.A{
bson.M{"$eq": bson.A{"$status", tender.TenderStatusActive}}, 1, 0,
}}},
"awarded": bson.M{"$sum": bson.M{"$cond": bson.A{
bson.M{"$eq": bson.A{"$status", tender.TenderStatusAwarded}}, 1, 0,
}}},
"cancelled": bson.M{"$sum": bson.M{"$cond": bson.A{
bson.M{"$eq": bson.A{"$status", tender.TenderStatusCancelled}}, 1, 0,
}}},
"expired": bson.M{"$sum": bson.M{"$cond": bson.A{
bson.M{"$and": bson.A{
bson.M{"$not": bson.M{"$in": bson.A{
"$status",
bson.A{tender.TenderStatusAwarded, tender.TenderStatusCancelled},
}}},
bson.M{"$or": bson.A{
bson.M{"$eq": bson.A{"$status", tender.TenderStatusExpired}},
bson.M{"$and": bson.A{
bson.M{"$gt": bson.A{"$expiry_deadline", 0}},
bson.M{"$lte": bson.A{"$expiry_deadline", now}},
}},
"$or": []bson.M{
{"status": tender.TenderStatusExpired},
{"expiry_deadline": bson.M{"$gt": 0, "$lte": now}},
},
}},
bson.M{"$count": "count"},
},
"closing_soon": bson.A{
bson.M{"$addFields": bson.M{"effective_deadline": effectiveDeadlineExpr()}},
bson.M{"$match": bson.M{
"effective_deadline": bson.M{"$gt": now, "$lte": windowEnd},
}},
bson.M{"$count": "count"},
},
"currency_value": bson.A{
bson.M{"$match": bson.M{
"estimated_value": bson.M{"$gt": 0},
"currency": bson.M{"$type": "string", "$ne": ""},
}},
bson.M{"$group": bson.M{
"_id": "$currency",
"count": bson.M{"$sum": 1},
"total": bson.M{"$sum": "$estimated_value"},
}},
bson.M{"$sort": bson.M{"count": -1}},
bson.M{"$limit": 1},
},
}}, 1, 0,
}}},
"closing_soon": bson.M{"$sum": bson.M{"$cond": bson.A{
bson.M{"$and": bson.A{
bson.M{"$gt": bson.A{"$effective_deadline", now}},
bson.M{"$lte": bson.A{"$effective_deadline", windowEnd}},
}}, 1, 0,
}}},
}}},
}
results, err := r.ormRepo.Aggregate(ctx, pipeline)
if err != nil {
return nil, fmt.Errorf("dashboard summary aggregation: %w", err)
return summaryStatusCounts{}, err
}
if len(results) == 0 {
return &SummaryResponse{
ValueCurrency: defaultValueCurrency,
GeneratedAt: now,
return summaryStatusCounts{}, nil
}
row := results[0]
return summaryStatusCounts{
total: aggregateCount(row, "total"),
active: aggregateCount(row, "active"),
awarded: aggregateCount(row, "awarded"),
expired: aggregateCount(row, "expired"),
cancelled: aggregateCount(row, "cancelled"),
closingSoon: aggregateCount(row, "closing_soon"),
}, nil
}
facet := results[0]
valueCurrency, totalValue := facetCurrencyValue(facet)
func (r *repository) summaryCurrencyValue(ctx context.Context) (string, float64, error) {
pipeline := mongo.Pipeline{
{{Key: "$match", Value: bson.M{
"estimated_value": bson.M{"$gt": 0},
"currency": bson.M{"$type": "string", "$ne": ""},
}}},
{{Key: "$group", Value: bson.M{
"_id": "$currency",
"count": bson.M{"$sum": 1},
"total": bson.M{"$sum": "$estimated_value"},
}}},
{{Key: "$sort", Value: bson.M{"count": -1}}},
{{Key: "$limit", Value: 1}},
}
return &SummaryResponse{
TotalTenders: facetCount(facet, "total"),
ActiveTenders: facetCount(facet, "active"),
AwardedTenders: facetCount(facet, "awarded"),
ExpiredTenders: facetCount(facet, "expired"),
CancelledTenders: facetCount(facet, "cancelled"),
ClosingSoon: facetCount(facet, "closing_soon"),
TotalEstimatedValue: totalValue,
ValueCurrency: valueCurrency,
GeneratedAt: now,
}, nil
results, err := r.ormRepo.Aggregate(ctx, pipeline)
if err != nil {
return defaultValueCurrency, 0, err
}
if len(results) == 0 {
return defaultValueCurrency, 0, nil
}
currency, total := facetCurrencyValue(bson.M{"currency_value": bson.A{results[0]}})
return currency, total, nil
}
func facetCount(facet bson.M, key string) int64 {
+163 -16
View File
@@ -22,7 +22,8 @@ const (
defaultCountriesLimit = 6
defaultListLimit = 5
maxListLimit = 20
summaryCacheTTL = 60 * time.Second
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
@@ -54,7 +55,7 @@ type service struct {
logger logger.Logger
redis redis.Client
summaryMu sync.Mutex
summary map[int64]cacheEntry[*SummaryResponse]
summary map[string]cacheEntry[*SummaryResponse]
summaryGroup singleflight.Group
statisticsMu sync.Mutex
statistics map[int]cacheEntry[*StatisticsReportResponse]
@@ -67,32 +68,46 @@ func NewService(repo Repository, log logger.Logger, redisClient redis.Client) Se
repo: repo,
logger: log,
redis: redisClient,
summary: make(map[int64]cacheEntry[*SummaryResponse]),
summary: make(map[string]cacheEntry[*SummaryResponse]),
statistics: make(map[int]cacheEntry[*StatisticsReportResponse]),
}
go s.warmStatisticsCache()
go s.warmSummaryCache()
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(windowSec); ok {
if cached, ok := s.cachedSummary(cacheKey); ok {
return cached, nil
}
cacheKey := strconv.FormatInt(windowSec, 10)
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(windowSec); ok {
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)
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(),
@@ -100,7 +115,8 @@ func (s *service) Summary(ctx context.Context, query SummaryQuery) (*SummaryResp
return nil, fmt.Errorf("dashboard summary: %w", err)
}
s.storeSummary(windowSec, result)
s.storeSummary(cacheKey, result)
s.storeRedisSummary(context.Background(), cacheKey, result)
return result, nil
})
if err != nil {
@@ -279,7 +295,6 @@ func emptyStatisticsReport(days int) *StatisticsReportResponse {
Days: days,
GeneratedAt: time.Now().UTC().Unix(),
Daily: StatisticsDailySeries{
ScrapedTED: fillTrendSeries(days, nil),
ScrapedDocuments: fillTrendSeries(days, nil),
TranslatedNotices: fillTrendSeries(days, nil),
},
@@ -324,16 +339,94 @@ func (s *service) loadStatistics(ctx context.Context, days int) (*StatisticsRepo
return out.(*StatisticsReportResponse), nil
}
func (s *service) cachedSummary(windowSec int64) (*SummaryResponse, bool) {
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[windowSec]
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, windowSec)
delete(s.summary, cacheKey)
}
return nil, false
}
@@ -341,17 +434,71 @@ func (s *service) cachedSummary(windowSec int64) (*SummaryResponse, bool) {
return entry.value, true
}
func (s *service) storeSummary(windowSec int64, value *SummaryResponse) {
func (s *service) storeSummary(cacheKey string, value *SummaryResponse) {
now := time.Now()
s.summaryMu.Lock()
defer s.summaryMu.Unlock()
s.summary[windowSec] = cacheEntry[*SummaryResponse]{
expiresAt: time.Now().Add(summaryCacheTTL),
staleUntil: time.Now().Add(summaryCacheTTL),
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()
@@ -41,7 +41,6 @@ func (r *repository) Statistics(ctx context.Context, days int, startUnix int64)
startDay := endDay.AddDate(0, 0, -(days - 1))
var (
scrapedTED map[string]int64
scrapedDocuments map[string]int64
translatedNotices map[string]int64
totalDocuments int64
@@ -62,24 +61,6 @@ func (r *repository) Statistics(ctx context.Context, days int, startUnix int64)
})
}
wg.Add(1)
go func() {
defer wg.Done()
qctx, cancel := context.WithTimeout(context.Background(), statisticsMongoQueryTimeout)
defer cancel()
// Read from the persistent metrics counter rather than the "notices" collection: the
// notice worker deletes processed notices shortly after promoting them into tenders, so
// counting from "notices" directly would undercount (often to zero) recent scrape activity.
counts, err := r.metricsCounter.GetTEDNoticeScrapedDailyCounts(qctx, startDay, endDay)
if err != nil {
recordFailure("scraped_ted", err)
scrapedTED = map[string]int64{}
return
}
scrapedTED = counts
}()
wg.Add(1)
go func() {
defer wg.Done()
@@ -172,7 +153,6 @@ func (r *repository) Statistics(ctx context.Context, days int, startUnix int64)
Days: days,
GeneratedAt: now.Unix(),
Daily: StatisticsDailySeries{
ScrapedTED: fillTrendSeries(days, scrapedTED),
ScrapedDocuments: fillTrendSeries(days, scrapedDocuments),
TranslatedNotices: fillTrendSeries(days, translatedNotices),
},