tm_back/internal/dashboard/statistics_repository.go

package dashboard

import (
	"context"
	"strings"
	"sync"
	"time"

	"tm/pkg/ai_summarizer"
	orm "tm/pkg/mongo"

	"go.mongodb.org/mongo-driver/v2/bson"
	mongodriver "go.mongodb.org/mongo-driver/v2/mongo"
)

// 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 = 15 * time.Minute

const (
	scrapedScopeResolveTimeout  = 3 * time.Minute
	statisticsMongoQueryTimeout = 90 * time.Second
	maxScrapedFolderInClause    = 5000
)

// 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 {
	match            bson.M
	zero             bool
	fromMinIO        bool
	dailyCounts      map[string]int64 // non-nil when scope was resolved from MinIO
	totalTenderCount int64            // >0 when MinIO procedure count should be used instead of Mongo Count
}

func (r *repository) Statistics(ctx context.Context, days int, startUnix int64) (*StatisticsReportResponse, error) {
	_ = ctx

	now := time.Now().UTC()
	endDay := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC)
	startDay := endDay.AddDate(0, 0, -(days - 1))

	var (
		scrapedTED        map[string]int64
		scrapedDocuments  map[string]int64
		translatedNotices map[string]int64
		totalDocuments    int64
		totalTranslated   int64
		totalScrapedTED   int64
		scrapedScope      scrapedTendersScope
	)

	var wg sync.WaitGroup
	var mu sync.Mutex

	recordFailure := func(section string, err error) {
		mu.Lock()
		defer mu.Unlock()
		r.logger.Warn("Dashboard statistics section failed", map[string]interface{}{
			"section": section,
			"error":   err.Error(),
		})
	}

	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()
		qctx, cancel := context.WithTimeout(context.Background(), statisticsMongoQueryTimeout)
		defer cancel()

		total, err := r.metricsCounter.Get(qctx, orm.TEDNoticeScrapedCounterKey)
		if err != nil {
			recordFailure("scraped_ted_total", err)
			return
		}
		totalScrapedTED = total
	}()

	wg.Add(1)
	go func() {
		defer wg.Done()
		qctx, cancel := context.WithTimeout(context.Background(), statisticsMongoQueryTimeout)
		defer cancel()

		counts, err := r.translatedNoticesPerDay(qctx, startDay, endDay)
		if err != nil {
			recordFailure("translated_notices", err)
			translatedNotices = map[string]int64{}
			return
		}
		translatedNotices = counts
	}()

	wg.Add(1)
	go func() {
		defer wg.Done()
		qctx, cancel := context.WithTimeout(context.Background(), statisticsMongoQueryTimeout)
		defer cancel()

		total, err := r.metricsCounter.Get(qctx, orm.AITranslationSuccessCounterKey)
		if err != nil {
			recordFailure("translated_total", err)
			return
		}
		totalTranslated = total
	}()

	wg.Add(1)
	go func() {
		defer wg.Done()
		scope, err := r.cachedScrapedTendersScope(context.Background())
		if err != nil {
			recordFailure("scraped_scope", err)
			scrapedScope = mongoScrapedTendersScope()
			return
		}
		scrapedScope = scope
	}()

	wg.Wait()

	wg.Add(1)
	go func() {
		defer wg.Done()
		qctx, cancel := context.WithTimeout(context.Background(), statisticsMongoQueryTimeout)
		defer cancel()

		counts, err := r.scrapedDocumentsPerDay(qctx, startUnix, scrapedScope)
		if err != nil {
			recordFailure("scraped_documents", err)
			scrapedDocuments = map[string]int64{}
			return
		}
		scrapedDocuments = counts
	}()

	wg.Add(1)
	go func() {
		defer wg.Done()
		qctx, cancel := context.WithTimeout(context.Background(), statisticsMongoQueryTimeout)
		defer cancel()

		total, err := r.totalScrapedTenders(qctx, scrapedScope)
		if err != nil {
			recordFailure("scraped_documents_total", err)
			return
		}
		totalDocuments = total
	}()

	wg.Wait()

	return &StatisticsReportResponse{
		Days:        days,
		GeneratedAt: now.Unix(),
		Daily: StatisticsDailySeries{
			ScrapedTED:        fillTrendSeries(days, scrapedTED),
			ScrapedDocuments:  fillTrendSeries(days, scrapedDocuments),
			TranslatedNotices: fillTrendSeries(days, translatedNotices),
		},
		Totals: StatisticsLifetimeTotals{
			ScrapedDocuments:  totalDocuments,
			TranslatedTenders: totalTranslated,
			ScrapedTEDNotices: totalScrapedTED,
		},
	}, nil
}

func (r *repository) scrapedDocumentsPerDay(ctx context.Context, startUnix int64, scope scrapedTendersScope) (map[string]int64, error) {
	if scope.zero {
		return map[string]int64{}, nil
	}

	if scope.dailyCounts != nil {
		return filterDailyCountsSince(scope.dailyCounts, startUnix), nil
	}

	return r.scrapedDocumentsPerDayFromMongo(ctx, startUnix, scope.match)
}

func filterDailyCountsSince(counts map[string]int64, startUnix int64) map[string]int64 {
	if len(counts) == 0 {
		return map[string]int64{}
	}
	startDay := time.Unix(startUnix, 0).UTC().Format("2006-01-02")
	filtered := make(map[string]int64)
	for date, count := range counts {
		if date >= startDay {
			filtered[date] = count
		}
	}
	return filtered
}

func (r *repository) scrapedDocumentsPerDayFromMongo(ctx context.Context, startUnix int64, match bson.M) (map[string]int64, error) {
	combinedMatch := bson.M{
		"processing_metadata.documents_scraped_at": bson.M{"$gte": startUnix, "$gt": 0},
	}
	for key, value := range match {
		combinedMatch[key] = value
	}

	pipeline := mongodriver.Pipeline{
		{{Key: "$match", Value: combinedMatch}},
		{{Key: "$addFields", Value: bson.M{
			"metric_ts": normalizeTimestampExpr("processing_metadata.documents_scraped_at"),
		}}},
		{{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},
		}}},
	}

	cursor, err := r.mongoManager.GetCollection(tenderCollectionName()).Aggregate(ctx, pipeline)
	if err != nil {
		return nil, err
	}
	defer cursor.Close(ctx)

	return decodeDailyCounts(ctx, cursor)
}

func (r *repository) translatedNoticesPerDay(ctx context.Context, startDay, endDay time.Time) (map[string]int64, error) {
	return r.metricsCounter.GetDailyCounts(ctx, startDay, endDay)
}

func (r *repository) totalScrapedTenders(ctx context.Context, scope scrapedTendersScope) (int64, error) {
	if scope.zero {
		return 0, nil
	}
	if scope.totalTenderCount > 0 {
		return scope.totalTenderCount, nil
	}

	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(_ 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()

		resolveCtx, cancel := context.WithTimeout(context.Background(), scrapedScopeResolveTimeout)
		defer cancel()

		scope, resolveErr := r.resolveScrapedTendersScope(resolveCtx)
		if resolveErr != nil {
			if r.logger != 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) {
	procedures, dailyCounts, err := r.listScrapedProcedures(ctx)
	if err != nil {
		return scrapedTendersScope{}, err
	}

	folderIDs := folderIDsFromProcedures(procedures)
	if len(folderIDs) > 0 {
		totalTenders := int64(len(folderIDs))
		if len(folderIDs) > maxScrapedFolderInClause {
			if r.logger != nil {
				r.logger.Warn("Too many MinIO procedure folders for dashboard statistics, using MinIO counts without Mongo $in filter", map[string]interface{}{
					"folder_count":  len(folderIDs),
					"max_in_clause": maxScrapedFolderInClause,
				})
			}
			return scrapedTendersScope{
				fromMinIO:        true,
				dailyCounts:      dailyCounts,
				totalTenderCount: totalTenders,
			}, nil
		}
		return scrapedTendersScope{
			match:            bson.M{"contract_folder_id": bson.M{"$in": folderIDs}},
			fromMinIO:        true,
			dailyCounts:      dailyCounts,
			totalTenderCount: totalTenders,
		}, nil
	}
	return mongoScrapedTendersScope(), nil
}

func (r *repository) listScrapedProcedures(ctx context.Context) ([]ai_summarizer.ProcedureDocumentsSummary, map[string]int64, error) {
	if r.procedureLister == nil {
		return nil, nil, nil
	}

	if scanner, ok := r.procedureLister.(scrapedDocumentsScanner); ok {
		return scanner.ScanScrapedDocuments(ctx)
	}

	procedures, err := r.procedureLister.ListProceduresWithDocuments(ctx)
	if err != nil {
		return nil, nil, err
	}
	return procedures, nil, nil
}

func folderIDsFromProcedures(procedures []ai_summarizer.ProcedureDocumentsSummary) []string {
	folderIDs := make([]string, 0, len(procedures))
	seen := make(map[string]struct{}, len(procedures))
	for _, proc := range procedures {
		if proc.DocumentCount <= 0 {
			continue
		}
		id := strings.TrimSpace(proc.ContractFolderID)
		if id == "" {
			continue
		}
		if _, ok := seen[id]; ok {
			continue
		}
		seen[id] = struct{}{}
		folderIDs = append(folderIDs, id)
	}
	return folderIDs
}

func (r *repository) scrapedDocumentsFolderIDs(ctx context.Context) ([]string, error) {
	if r.procedureLister == nil {
		return nil, nil
	}

	procedures, _, err := r.listScrapedProcedures(ctx)
	if err != nil {
		return nil, err
	}
	return folderIDsFromProcedures(procedures), nil
}

func decodeDailyCounts(ctx context.Context, cursor *mongodriver.Cursor) (map[string]int64, error) {
	counts := make(map[string]int64)
	for cursor.Next(ctx) {
		var row bson.M
		if err := cursor.Decode(&row); err != nil {
			return nil, err
		}
		date, ok := row["_id"].(string)
		if !ok || date == "" {
			continue
		}
		counts[date] = aggregateCount(row, "count")
	}
	if err := cursor.Err(); err != nil {
		return nil, err
	}
	return counts, nil
}