tm_back/internal/dashboard/service.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
	// statisticsColdLoadWait bounds how long a cold-cache request waits for a fresh
	// load before falling back to a placeholder. Mongo-only loads (MinIO scope
	// already cached) comfortably finish within this window; a first-ever MinIO
	// bucket scan does not, so we don't block the request for that.
	statisticsColdLoadWait = 4 * time.Second
)

// 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
	}

	// Cold cache: the underlying load performs a full MinIO bucket scan plus Mongo
	// aggregations that can take minutes on a first run. Start the load in the
	// background (deduped via singleflight, and detached from this request's
	// context so it keeps running even if we give up waiting) and wait briefly for
	// it. If it finishes in time we return real data; otherwise we serve a
	// placeholder so the endpoint stays fast, and the load keeps warming the cache
	// for the next request.
	resultCh := make(chan *StatisticsReportResponse, 1)
	go func() {
		result, err := s.loadStatistics(context.Background(), days)
		if err != nil {
			resultCh <- nil
			return
		}
		resultCh <- result
	}()

	select {
	case result := <-resultCh:
		if result != nil {
			return result, nil
		}
	case <-time.After(statisticsColdLoadWait):
		s.logger.Info("Serving placeholder dashboard statistics while cache warms", map[string]interface{}{
			"days": days,
		})
	}

	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
}