Mazyar
81b0d94ba3
- Introduced a new `widgetCache` structure to manage caching for various dashboard widgets, improving performance and reducing redundant data retrieval. - Updated the `ClosingSoon` and `Trend` methods in the service to utilize the new caching mechanism, ensuring efficient data access and reducing load on the database. - Enhanced the `NoticeTypes` and `Countries` methods to implement caching, further optimizing the dashboard's responsiveness. - Added a new `searchListCache` implementation in the tender service to cache search results, improving the efficiency of tender searches and reducing database load. - Implemented robust error handling and logging for cache operations, ensuring better visibility into potential issues. This update significantly enhances the performance and scalability of the dashboard and tender services by integrating effective caching strategies, leading to improved user experience and resource management.
716 lines
18 KiB
Go
716 lines
18 KiB
Go
package dashboard
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
"tm/internal/tender"
|
|
"tm/pkg/ai_summarizer"
|
|
"tm/pkg/logger"
|
|
orm "tm/pkg/mongo"
|
|
|
|
"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.
|
|
// This is the source of truth for document-scrape statistics, matching tender panel search.
|
|
type ProcedureDocumentsLister interface {
|
|
ListProceduresWithDocuments(ctx context.Context) ([]ai_summarizer.ProcedureDocumentsSummary, error)
|
|
}
|
|
|
|
// scrapedDocumentsScanner optionally scans MinIO for per-day document counts in the same pass.
|
|
type scrapedDocumentsScanner interface {
|
|
ScanScrapedDocuments(ctx context.Context) ([]ai_summarizer.ProcedureDocumentsSummary, map[string]int64, error)
|
|
}
|
|
|
|
// translatedNoticesScanner optionally scans MinIO for per-day translated notice counts.
|
|
type translatedNoticesScanner interface {
|
|
ScanTranslatedNotices(ctx context.Context) (map[string]int64, int64, error)
|
|
}
|
|
|
|
const defaultValueCurrency = "EUR"
|
|
|
|
// Repository defines dashboard data access.
|
|
type Repository interface {
|
|
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)
|
|
ClosingSoon(ctx context.Context, limit int, windowSec int64) ([]ClosingSoonItem, error)
|
|
Recent(ctx context.Context, limit int, cursor string) ([]RecentItem, *string, error)
|
|
Statistics(ctx context.Context, days int, startUnix int64) (*StatisticsReportResponse, error)
|
|
}
|
|
|
|
type repository struct {
|
|
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
|
|
translatedScopeMu sync.Mutex
|
|
translatedScope translatedNoticesScope
|
|
translatedScopeExpiry time.Time
|
|
translatedScopeCached bool
|
|
translatedScopeGroup singleflight.Group
|
|
}
|
|
|
|
// NewRepository creates a dashboard repository backed by the tenders collection.
|
|
func NewRepository(mongoManager *orm.ConnectionManager, log logger.Logger, procedureLister ProcedureDocumentsLister) Repository {
|
|
return &repository{
|
|
ormRepo: orm.NewRepository[tender.Tender](mongoManager.GetCollection(tenderCollectionName()), log),
|
|
mongoManager: mongoManager,
|
|
metricsCounter: orm.NewCounter(mongoManager),
|
|
procedureLister: procedureLister,
|
|
logger: log,
|
|
}
|
|
}
|
|
|
|
func tenderCollectionName() string {
|
|
return "tenders"
|
|
}
|
|
|
|
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: "$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}},
|
|
}},
|
|
}},
|
|
}}, 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 summaryStatusCounts{}, err
|
|
}
|
|
if len(results) == 0 {
|
|
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
|
|
}
|
|
|
|
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}},
|
|
}
|
|
|
|
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 {
|
|
rows := facetRows(facet, key)
|
|
if len(rows) == 0 {
|
|
return 0
|
|
}
|
|
return aggregateCount(asBSONMap(rows[0]), "count")
|
|
}
|
|
|
|
func facetCurrencyValue(facet bson.M) (string, float64) {
|
|
rows := facetRows(facet, "currency_value")
|
|
if len(rows) == 0 {
|
|
return defaultValueCurrency, 0
|
|
}
|
|
row := asBSONMap(rows[0])
|
|
if row == nil {
|
|
return defaultValueCurrency, 0
|
|
}
|
|
|
|
currency := defaultValueCurrency
|
|
if c, ok := row["_id"].(string); ok && c != "" {
|
|
currency = strings.ToUpper(c)
|
|
}
|
|
return currency, toFloat64(row["total"])
|
|
}
|
|
|
|
func facetRows(facet bson.M, key string) []interface{} {
|
|
raw, ok := facet[key]
|
|
if !ok {
|
|
return nil
|
|
}
|
|
switch rows := raw.(type) {
|
|
case bson.A:
|
|
return []interface{}(rows)
|
|
case []interface{}:
|
|
return rows
|
|
default:
|
|
return nil
|
|
}
|
|
}
|
|
|
|
func asBSONMap(v interface{}) bson.M {
|
|
switch doc := v.(type) {
|
|
case bson.M:
|
|
return doc
|
|
case bson.D:
|
|
out := make(bson.M, len(doc))
|
|
for _, elem := range doc {
|
|
out[elem.Key] = elem.Value
|
|
}
|
|
return out
|
|
case map[string]interface{}:
|
|
return doc
|
|
default:
|
|
return nil
|
|
}
|
|
}
|
|
|
|
func (r *repository) Trend(ctx context.Context, days int, metric string, startUnix int64) (map[string]int64, error) {
|
|
field := trendTimestampField(metric)
|
|
|
|
pipeline := mongo.Pipeline{
|
|
{{Key: "$match", Value: bson.M{
|
|
field: bson.M{"$gte": startUnix, "$gt": 0},
|
|
}}},
|
|
{{Key: "$addFields", Value: bson.M{
|
|
"metric_ts": normalizeTimestampExpr(field),
|
|
}}},
|
|
{{Key: "$group", Value: bson.M{
|
|
"_id": bson.M{
|
|
"$dateToString": bson.M{
|
|
"format": "%Y-%m-%d",
|
|
"date": bson.M{"$toDate": bson.M{"$multiply": bson.A{"$metric_ts", 1000}}},
|
|
"timezone": "UTC",
|
|
},
|
|
},
|
|
"count": bson.M{"$sum": 1},
|
|
}}},
|
|
}
|
|
|
|
results, err := r.ormRepo.Aggregate(ctx, pipeline)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("trend aggregation: %w", err)
|
|
}
|
|
|
|
counts := make(map[string]int64, len(results))
|
|
for _, row := range results {
|
|
date, ok := row["_id"].(string)
|
|
if !ok || date == "" {
|
|
continue
|
|
}
|
|
counts[date] = aggregateCount(row, "count")
|
|
}
|
|
|
|
return counts, nil
|
|
}
|
|
|
|
func (r *repository) Countries(ctx context.Context, limit int) (*CountriesResponse, error) {
|
|
pipeline := mongo.Pipeline{
|
|
{{Key: "$match", Value: bson.M{
|
|
"country_code": bson.M{"$type": "string", "$ne": ""},
|
|
}}},
|
|
{{Key: "$group", Value: bson.M{
|
|
"_id": bson.M{"$toUpper": "$country_code"},
|
|
"count": bson.M{"$sum": 1},
|
|
}}},
|
|
{{Key: "$sort", Value: bson.M{"count": -1}}},
|
|
}
|
|
|
|
results, err := r.ormRepo.Aggregate(ctx, pipeline)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("countries aggregation: %w", err)
|
|
}
|
|
|
|
var total int64
|
|
items := make([]CountryItem, 0, limit)
|
|
var other int64
|
|
|
|
for i, row := range results {
|
|
code, _ := row["_id"].(string)
|
|
count := aggregateCount(row, "count")
|
|
total += count
|
|
|
|
if code == "" {
|
|
other += count
|
|
continue
|
|
}
|
|
|
|
if i < limit {
|
|
items = append(items, CountryItem{CountryCode: code, Count: count})
|
|
} else {
|
|
other += count
|
|
}
|
|
}
|
|
|
|
return &CountriesResponse{
|
|
Total: total,
|
|
Items: items,
|
|
OtherCount: other,
|
|
}, nil
|
|
}
|
|
|
|
func (r *repository) NoticeTypes(ctx context.Context) (*NoticeTypesResponse, error) {
|
|
pipeline := mongo.Pipeline{
|
|
{{Key: "$group", Value: bson.M{
|
|
"_id": bson.M{
|
|
"$cond": bson.A{
|
|
bson.M{"$and": bson.A{
|
|
bson.M{"$ne": bson.A{"$notice_type_code", ""}},
|
|
bson.M{"$ne": bson.A{"$notice_type_code", nil}},
|
|
}},
|
|
"$notice_type_code",
|
|
"UNKNOWN",
|
|
},
|
|
},
|
|
"count": bson.M{"$sum": 1},
|
|
}}},
|
|
{{Key: "$sort", Value: bson.M{"count": -1}}},
|
|
}
|
|
|
|
results, err := r.ormRepo.Aggregate(ctx, pipeline)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("notice types aggregation: %w", err)
|
|
}
|
|
|
|
items := make([]NoticeTypeItem, 0, len(results))
|
|
var total int64
|
|
for _, row := range results {
|
|
noticeType, _ := row["_id"].(string)
|
|
count := aggregateCount(row, "count")
|
|
total += count
|
|
items = append(items, NoticeTypeItem{Type: noticeType, Count: count})
|
|
}
|
|
|
|
return &NoticeTypesResponse{Total: total, Items: items}, nil
|
|
}
|
|
|
|
func (r *repository) ClosingSoon(ctx context.Context, limit int, windowSec int64) ([]ClosingSoonItem, error) {
|
|
now := time.Now().Unix()
|
|
windowEnd := now + windowSec
|
|
deadlineRange := bson.M{"$gt": now, "$lte": windowEnd}
|
|
|
|
pipeline := mongo.Pipeline{
|
|
{{Key: "$match", Value: bson.M{
|
|
"$or": bson.A{
|
|
bson.M{"submission_deadline": deadlineRange},
|
|
bson.M{
|
|
"$and": bson.A{
|
|
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}},
|
|
}},
|
|
bson.M{"tender_deadline": deadlineRange},
|
|
},
|
|
},
|
|
},
|
|
}}},
|
|
{{Key: "$addFields", Value: bson.M{"effective_deadline": effectiveDeadlineExpr()}}},
|
|
{{Key: "$sort", Value: bson.M{"effective_deadline": 1}}},
|
|
{{Key: "$limit", Value: limit}},
|
|
{{Key: "$project", Value: bson.M{
|
|
"_id": 1,
|
|
"title": 1,
|
|
"country_code": 1,
|
|
"buyer_organization": 1,
|
|
"submission_deadline": 1,
|
|
"tender_deadline": 1,
|
|
"status": 1,
|
|
"estimated_value": 1,
|
|
"currency": 1,
|
|
"effective_deadline": 1,
|
|
}}},
|
|
}
|
|
|
|
results, err := r.ormRepo.Aggregate(ctx, pipeline)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("closing soon list: %w", err)
|
|
}
|
|
|
|
items := make([]ClosingSoonItem, 0, len(results))
|
|
for _, row := range results {
|
|
items = append(items, mapClosingSoonRow(row))
|
|
}
|
|
|
|
return items, nil
|
|
}
|
|
|
|
func (r *repository) Recent(ctx context.Context, limit int, cursor string) ([]RecentItem, *string, error) {
|
|
filter := bson.M{}
|
|
|
|
mongoPagination, err := orm.BuildListPagination(
|
|
limit,
|
|
0,
|
|
cursor,
|
|
"created_at",
|
|
"desc",
|
|
filter,
|
|
orm.ListPaginationOptions{
|
|
Projection: bson.M{
|
|
"title": 1,
|
|
"country_code": 1,
|
|
"status": 1,
|
|
"buyer_organization": 1,
|
|
"created_at": 1,
|
|
"publication_date": 1,
|
|
},
|
|
},
|
|
)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("recent pagination: %w", err)
|
|
}
|
|
|
|
result, err := r.ormRepo.FindAll(ctx, filter, mongoPagination)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("recent list: %w", err)
|
|
}
|
|
|
|
items := make([]RecentItem, 0, len(result.Items))
|
|
for _, t := range result.Items {
|
|
items = append(items, mapRecentTender(&t))
|
|
}
|
|
|
|
var nextCursor *string
|
|
if result.NextCursor != "" {
|
|
nextCursor = &result.NextCursor
|
|
}
|
|
|
|
return items, nextCursor, nil
|
|
}
|
|
|
|
func mapClosingSoonRow(row bson.M) ClosingSoonItem {
|
|
item := ClosingSoonItem{
|
|
Title: stringField(row, "title"),
|
|
CountryCode: strings.ToUpper(stringField(row, "country_code")),
|
|
Status: stringField(row, "status"),
|
|
EstimatedValue: toFloat64(row["estimated_value"]),
|
|
Currency: strings.ToUpper(stringField(row, "currency")),
|
|
}
|
|
|
|
if id, ok := row["_id"]; ok {
|
|
item.ID = objectIDString(id)
|
|
}
|
|
|
|
item.SubmissionDeadline = normalizeUnixTimestamp(int64Field(row, "submission_deadline"))
|
|
item.TenderDeadline = normalizeUnixTimestamp(int64Field(row, "tender_deadline"))
|
|
|
|
if effective := int64Field(row, "effective_deadline"); effective > 0 && item.SubmissionDeadline == 0 {
|
|
item.SubmissionDeadline = effective
|
|
}
|
|
|
|
if buyer, ok := row["buyer_organization"].(bson.M); ok {
|
|
item.BuyerName = stringField(buyer, "name")
|
|
} else if buyer, ok := row["buyer_organization"].(map[string]interface{}); ok {
|
|
if name, ok := buyer["name"].(string); ok {
|
|
item.BuyerName = name
|
|
}
|
|
}
|
|
|
|
return item
|
|
}
|
|
|
|
func mapRecentTender(t *tender.Tender) RecentItem {
|
|
item := RecentItem{
|
|
ID: t.GetID(),
|
|
Title: t.Title,
|
|
CountryCode: strings.ToUpper(t.CountryCode),
|
|
Status: string(t.Status),
|
|
CreatedAt: normalizeUnixTimestamp(t.CreatedAt),
|
|
PublicationDate: normalizeUnixTimestamp(t.PublicationDate),
|
|
}
|
|
|
|
if t.BuyerOrganization != nil {
|
|
item.BuyerName = t.BuyerOrganization.Name
|
|
}
|
|
|
|
if item.CreatedAt == 0 && item.PublicationDate > 0 {
|
|
item.CreatedAt = item.PublicationDate
|
|
}
|
|
|
|
return item
|
|
}
|
|
|
|
func trendTimestampField(metric string) string {
|
|
switch strings.ToLower(strings.TrimSpace(metric)) {
|
|
case "published":
|
|
return "publication_date"
|
|
case "awarded":
|
|
return "award_date"
|
|
default:
|
|
return "created_at"
|
|
}
|
|
}
|
|
|
|
// effectiveDeadlineExpr is used for closing-soon (submission first, per dashboard-api.md).
|
|
func effectiveDeadlineExpr() bson.M {
|
|
return normalizeDeadlineFieldExpr(
|
|
"$submission_deadline",
|
|
"$tender_deadline",
|
|
)
|
|
}
|
|
|
|
// expiryDeadlineExpr is used for expired counts (tender deadline takes precedence).
|
|
func expiryDeadlineExpr() bson.M {
|
|
return normalizeDeadlineFieldExpr(
|
|
"$tender_deadline",
|
|
"$submission_deadline",
|
|
)
|
|
}
|
|
|
|
func normalizeDeadlineFieldExpr(primaryField, fallbackField string) bson.M {
|
|
return bson.M{
|
|
"$let": bson.M{
|
|
"vars": bson.M{
|
|
"raw": bson.M{
|
|
"$cond": bson.A{
|
|
bson.M{"$gt": bson.A{primaryField, 0}},
|
|
primaryField,
|
|
fallbackField,
|
|
},
|
|
},
|
|
},
|
|
"in": normalizeTimestampExpr("$$raw"),
|
|
},
|
|
}
|
|
}
|
|
|
|
func normalizeTimestampExpr(field string) bson.M {
|
|
ref := field
|
|
if !strings.HasPrefix(ref, "$") {
|
|
ref = "$" + ref
|
|
}
|
|
return bson.M{
|
|
"$cond": bson.A{
|
|
bson.M{"$gt": bson.A{ref, int64(1_000_000_000_000)}},
|
|
bson.M{"$divide": bson.A{ref, 1000}},
|
|
ref,
|
|
},
|
|
}
|
|
}
|
|
|
|
func aggregateCount(row bson.M, field string) int64 {
|
|
if row == nil {
|
|
return 0
|
|
}
|
|
if field != "" {
|
|
if v, ok := row[field]; ok {
|
|
return toInt64(v)
|
|
}
|
|
}
|
|
if v, ok := row["count"]; ok {
|
|
return toInt64(v)
|
|
}
|
|
return 0
|
|
}
|
|
|
|
func toInt64(v interface{}) int64 {
|
|
switch n := v.(type) {
|
|
case int32:
|
|
return int64(n)
|
|
case int64:
|
|
return n
|
|
case float64:
|
|
return int64(n)
|
|
case int:
|
|
return int64(n)
|
|
default:
|
|
return 0
|
|
}
|
|
}
|
|
|
|
func toFloat64(v interface{}) float64 {
|
|
switch n := v.(type) {
|
|
case float64:
|
|
return n
|
|
case float32:
|
|
return float64(n)
|
|
case int32:
|
|
return float64(n)
|
|
case int64:
|
|
return float64(n)
|
|
case int:
|
|
return float64(n)
|
|
default:
|
|
return 0
|
|
}
|
|
}
|
|
|
|
func int64Field(row bson.M, key string) int64 {
|
|
return toInt64(row[key])
|
|
}
|
|
|
|
func stringField(row bson.M, key string) string {
|
|
if v, ok := row[key].(string); ok {
|
|
return v
|
|
}
|
|
return ""
|
|
}
|
|
|
|
func objectIDString(v interface{}) string {
|
|
switch id := v.(type) {
|
|
case bson.ObjectID:
|
|
return id.Hex()
|
|
default:
|
|
return fmt.Sprint(v)
|
|
}
|
|
}
|
|
|
|
func normalizeUnixTimestamp(value int64) int64 {
|
|
if value > 1_000_000_000_000 {
|
|
return value / 1000
|
|
}
|
|
return value
|
|
}
|