mazyar/tm_back
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'develop' into est-value

Browse Source
This commit is contained in:
m.nazemi
2026-06-21 09:37:00 +03:30
parent 4cfca5aa55 dfab3e17d2
commit a04b3fe1ec
4 changed files with 260 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/dashboard/repository.go
+97 -111
View File
@@ -4,6 +4,7 @@ import (
"context"
"fmt"
"strings"
"sync"
"time"
"tm/internal/tender"
"tm/pkg/ai_summarizer"
@@ -12,6 +13,7 @@ import (
"go.mongodb.org/mongo-driver/v2/bson"
"go.mongodb.org/mongo-driver/v2/mongo"
"golang.org/x/sync/singleflight"
)
// ProcedureDocumentsLister lists contract folders that have scraped documents in MinIO.
@@ -34,11 +36,16 @@ type Repository interface {
}
type repository struct {
ormRepo orm.Repository[tender.Tender]
mongoManager *orm.ConnectionManager
metricsCounter *orm.Counter
procedureLister ProcedureDocumentsLister
logger logger.Logger
ormRepo orm.Repository[tender.Tender]
mongoManager *orm.ConnectionManager
metricsCounter *orm.Counter
procedureLister ProcedureDocumentsLister
logger logger.Logger
scrapedScopeMu sync.Mutex
scrapedScope scrapedTendersScope
scrapedScopeExpiry time.Time
scrapedScopeCached bool
scrapedScopeGroup singleflight.Group
}
// NewRepository creates a dashboard repository backed by the tenders collection.
@@ -60,135 +67,114 @@ func (r *repository) Summary(ctx context.Context, closingWindowSec int64) (*Summ
now := time.Now().Unix()
windowEnd := now + closingWindowSec
total, err := r.ormRepo.Count(ctx, bson.M{})
if err != nil {
return nil, fmt.Errorf("count total tenders: %w", err)
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,
}},
"$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},
},
}}},
}
active, err := r.ormRepo.Count(ctx, bson.M{"status": tender.TenderStatusActive})
results, err := r.ormRepo.Aggregate(ctx, pipeline)
if err != nil {
return nil, fmt.Errorf("count active tenders: %w", err)
return nil, fmt.Errorf("dashboard summary aggregation: %w", err)
}
if len(results) == 0 {
return &SummaryResponse{
ValueCurrency: defaultValueCurrency,
GeneratedAt: now,
}, nil
}
awarded, err := r.ormRepo.Count(ctx, bson.M{"status": tender.TenderStatusAwarded})
if err != nil {
return nil, fmt.Errorf("count awarded tenders: %w", err)
}
expired, err := r.countExpired(ctx, now)
if err != nil {
return nil, err
}
cancelled, err := r.ormRepo.Count(ctx, bson.M{"status": tender.TenderStatusCancelled})
if err != nil {
return nil, fmt.Errorf("count cancelled tenders: %w", err)
}
closingSoon, err := r.countClosingSoon(ctx, now, windowEnd)
if err != nil {
return nil, err
}
valueCurrency, totalValue, err := r.dominantCurrencyValue(ctx)
if err != nil {
return nil, err
}
facet := results[0]
valueCurrency, totalValue := facetCurrencyValue(facet)
return &SummaryResponse{
TotalTenders: total,
ActiveTenders: active,
AwardedTenders: awarded,
ExpiredTenders: expired,
CancelledTenders: cancelled,
ClosingSoon: closingSoon,
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
}
func (r *repository) countExpired(ctx context.Context, now int64) (int64, error) {
pipeline := mongo.Pipeline{
{{Key: "$addFields", Value: bson.M{"expiry_deadline": expiryDeadlineExpr()}}},
{{Key: "$match", Value: bson.M{
"status": bson.M{"$nin": []tender.TenderStatus{
tender.TenderStatusAwarded,
tender.TenderStatusCancelled,
}},
"$or": []bson.M{
{"status": tender.TenderStatusExpired},
{
"expiry_deadline": bson.M{"$gt": 0, "$lte": now},
},
},
}}},
{{Key: "$count", Value: "count"}},
func facetCount(facet bson.M, key string) int64 {
rows, ok := facet[key].(bson.A)
if !ok || len(rows) == 0 {
return 0
}
results, err := r.ormRepo.Aggregate(ctx, pipeline)
if err != nil {
return 0, fmt.Errorf("count expired tenders: %w", err)
row, ok := rows[0].(bson.M)
if !ok {
return 0
}
if len(results) == 0 {
return 0, nil
}
return aggregateCount(results[0], "count"), nil
return aggregateCount(row, "count")
}
func (r *repository) countClosingSoon(ctx context.Context, now, windowEnd int64) (int64, error) {
pipeline := mongo.Pipeline{
{{Key: "$addFields", Value: bson.M{"effective_deadline": effectiveDeadlineExpr()}}},
{{Key: "$match", Value: bson.M{
"effective_deadline": bson.M{"$gt": now, "$lte": windowEnd},
}}},
{{Key: "$count", Value: "count"}},
func facetCurrencyValue(facet bson.M) (string, float64) {
rows, ok := facet["currency_value"].(bson.A)
if !ok || len(rows) == 0 {
return defaultValueCurrency, 0
}
results, err := r.ormRepo.Aggregate(ctx, pipeline)
if err != nil {
return 0, fmt.Errorf("count closing soon: %w", err)
}
if len(results) == 0 {
return 0, nil
}
return aggregateCount(results[0], "count"), nil
}
func (r *repository) dominantCurrencyValue(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}},
}
results, err := r.ormRepo.Aggregate(ctx, pipeline)
if err != nil {
return defaultValueCurrency, 0, fmt.Errorf("dominant currency value: %w", err)
}
if len(results) == 0 {
return defaultValueCurrency, 0, nil
row, ok := rows[0].(bson.M)
if !ok {
return defaultValueCurrency, 0
}
currency := defaultValueCurrency
if c, ok := results[0]["_id"].(string); ok && c != "" {
if c, ok := row["_id"].(string); ok && c != "" {
currency = strings.ToUpper(c)
}
return currency, toFloat64(results[0]["total"]), nil
return currency, toFloat64(row["total"])
}
func (r *repository) Trend(ctx context.Context, days int, metric string, startUnix int64) (map[string]int64, error) {
internal/dashboard/service.go
+95 -30
View File
@@ -3,10 +3,13 @@ package dashboard
import (
"context"
"fmt"
"strconv"
"strings"
"sync"
"time"
"tm/pkg/logger"
"golang.org/x/sync/singleflight"
)
const (
@@ -16,6 +19,7 @@ const (
defaultCountriesLimit = 6
defaultListLimit = 5
maxListLimit = 20
summaryCacheTTL = 60 * time.Second
statisticsCacheTTL = 60 * time.Second
)
@@ -30,16 +34,20 @@ type Service interface {
Statistics(ctx context.Context, query StatisticsQuery) (*StatisticsReportResponse, error)
}
type statisticsCacheEntry struct {
type cacheEntry[T any] struct {
expiresAt time.Time
value *StatisticsReportResponse
value T
}
type service struct {
repo Repository
logger logger.Logger
statisticsMu sync.Mutex
statistics map[int]statisticsCacheEntry
repo Repository
logger logger.Logger
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.
@@ -47,7 +55,8 @@ func NewService(repo Repository, log logger.Logger) Service {
return &service{
repo: repo,
logger: log,
statistics: make(map[int]statisticsCacheEntry),
summary: make(map[int64]cacheEntry[*SummaryResponse]),
statistics: make(map[int]cacheEntry[*StatisticsReportResponse]),
}
}
@@ -55,19 +64,36 @@ func (s *service) Summary(ctx context.Context, query SummaryQuery) (*SummaryResp
windowHours := closingWindowHours(query.ClosingWindow)
windowSec := int64(windowHours) * 3600
s.logger.Info("Fetching dashboard summary", map[string]interface{}{
"closing_window_hours": windowHours,
})
out, err := s.repo.Summary(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)
if cached, ok := s.cachedSummary(windowSec); ok {
return cached, nil
}
return out, 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(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) {
@@ -177,22 +203,61 @@ func (s *service) Statistics(ctx context.Context, query StatisticsQuery) (*Stati
return cached, nil
}
startUnix := trendStartUnix(days)
cacheKey := strconv.Itoa(days)
out, err, _ := s.statisticsGroup.Do(cacheKey, func() (interface{}, error) {
if cached, ok := s.cachedStatistics(days); ok {
return cached, nil
}
s.logger.Info("Fetching dashboard statistics report", map[string]interface{}{
"days": days,
})
startUnix := trendStartUnix(days)
out, err := s.repo.Statistics(ctx, days, startUnix)
if err != nil {
s.logger.Error("Failed to fetch dashboard statistics report", map[string]interface{}{
"error": err.Error(),
s.logger.Info("Fetching dashboard statistics report", map[string]interface{}{
"days": days,
})
return nil, fmt.Errorf("dashboard statistics: %w", err)
result, err := s.repo.Statistics(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)
return result, nil
})
if err != nil {
return nil, err
}
s.storeStatistics(days, out)
return out, nil
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),
value: value,
}
}
func (s *service) cachedStatistics(days int) (*StatisticsReportResponse, bool) {
@@ -216,7 +281,7 @@ func (s *service) storeStatistics(days int, value *StatisticsReportResponse) {
s.statisticsMu.Lock()
defer s.statisticsMu.Unlock()
s.statistics[days] = statisticsCacheEntry{
s.statistics[days] = cacheEntry[*StatisticsReportResponse]{
expiresAt: time.Now().Add(statisticsCacheTTL),
value: value,
}
internal/dashboard/statistics_repository.go
+58 -1
View File
@@ -16,6 +16,10 @@ import (
const noticesCollectionName = "notices"
// scrapedScopeCacheTTL avoids listing every MinIO object on each statistics request while
// keeping counts aligned with the tender panel (MinIO is the source of truth for scraped docs).
const scrapedScopeCacheTTL = 5 * time.Minute
// scrapedTendersScope is the resolved Mongo filter for tenders with scraped documents.
// zero is set when MinIO reports no procedure folders; callers must return zero without querying.
type scrapedTendersScope struct {
@@ -28,7 +32,8 @@ func (r *repository) Statistics(ctx context.Context, days int, startUnix int64)
endDay := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC)
startDay := endDay.AddDate(0, 0, -(days - 1))
scrapedScope, err := r.resolveScrapedTendersScope(ctx)
// Resolve scraped-document scope from MinIO (cached); falls back to Mongo flags on error.
scrapedScope, err := r.cachedScrapedTendersScope(ctx)
if err != nil {
return nil, err
}
@@ -187,6 +192,58 @@ func (r *repository) totalScrapedTenders(ctx context.Context, scope scrapedTende
return r.ormRepo.Count(ctx, scope.match)
}
func mongoScrapedTendersScope() scrapedTendersScope {
return scrapedTendersScope{
match: bson.M{
"processing_metadata.documents_scraped": true,
"scraped_documents": bson.M{"$exists": true, "$ne": bson.A{}},
},
}
}
func (r *repository) cachedScrapedTendersScope(ctx context.Context) (scrapedTendersScope, error) {
now := time.Now()
r.scrapedScopeMu.Lock()
if r.scrapedScopeCached && now.Before(r.scrapedScopeExpiry) {
scope := r.scrapedScope
r.scrapedScopeMu.Unlock()
return scope, nil
}
r.scrapedScopeMu.Unlock()
v, err, _ := r.scrapedScopeGroup.Do("scraped-scope", func() (interface{}, error) {
r.scrapedScopeMu.Lock()
if r.scrapedScopeCached && time.Now().Before(r.scrapedScopeExpiry) {
scope := r.scrapedScope
r.scrapedScopeMu.Unlock()
return scope, nil
}
r.scrapedScopeMu.Unlock()
scope, resolveErr := r.resolveScrapedTendersScope(ctx)
if resolveErr != nil {
r.logger.Warn("MinIO scraped scope unavailable, using Mongo fallback for dashboard statistics", map[string]interface{}{
"error": resolveErr.Error(),
})
return mongoScrapedTendersScope(), nil
}
r.scrapedScopeMu.Lock()
r.scrapedScope = scope
r.scrapedScopeExpiry = time.Now().Add(scrapedScopeCacheTTL)
r.scrapedScopeCached = true
r.scrapedScopeMu.Unlock()
return scope, nil
})
if err != nil {
return scrapedTendersScope{}, err
}
return v.(scrapedTendersScope), nil
}
func (r *repository) resolveScrapedTendersScope(ctx context.Context) (scrapedTendersScope, error) {
folderIDs, err := r.scrapedDocumentsFolderIDs(ctx)
if err != nil {
internal/dashboard/statistics_repository_test.go
+10
View File
@@ -129,3 +129,13 @@ func TestResolveScrapedTendersScopeMongoFallbackWithoutLister(t *testing.T) {
t.Fatalf("expected mongo fallback filter, got %#v", scope.match)
}
}
func TestMongoScrapedTendersScopeMatchesStatisticsFilter(t *testing.T) {
scope := mongoScrapedTendersScope()
if scope.zero {
t.Fatal("expected non-zero mongo statistics scope")
}
if scope.match["processing_metadata.documents_scraped"] != true {
t.Fatalf("expected documents_scraped filter, got %#v", scope.match)
}
}