gochat/backend/internal/service/reporting_rollup_service.go

package service

import (
	"context"
	"time"

	"github.com/gochat/gochat/internal/model"
	"github.com/gochat/gochat/internal/repository"
	applogger "github.com/gochat/gochat/pkg/logger"
)

// ReportingRollupService computes daily/weekly/monthly rollup aggregations
// from raw ReportingEvents. Reference: Chatwoot app/services/reporting_events/rollup_service.rb (81行)
//
// Rollup computation:
//  1. Group raw events by (account_id, date, dimension_type, dimension_id, metric)
//  2. Compute COUNT, SUM(value), SUM(value_in_business_hours) per group
//  3. Upsert into reporting_events_rollups table
type ReportingRollupService struct {
	eventRepo  *repository.ReportingEventRepo
	rollupRepo *repository.ReportingEventsRollupRepo
}

// NewReportingRollupService creates a new rollup service.
func NewReportingRollupService(eventRepo *repository.ReportingEventRepo, rollupRepo *repository.ReportingEventsRollupRepo) *ReportingRollupService {
	return &ReportingRollupService{eventRepo: eventRepo, rollupRepo: rollupRepo}
}

// RollupEvent additively applies a freshly saved reporting event to daily rollups.
// Chatwoot does this from ReportingEventListener#safe_rollup after the raw event is saved.
func (s *ReportingRollupService) RollupEvent(ctx context.Context, event *model.ReportingEvent) error {
	if event == nil || s == nil || s.rollupRepo == nil {
		return nil
	}
	loc, enabled, err := accountReportingLocation(ctx, reportingServiceDB(s.eventRepo, s.rollupRepo), event.AccountID)
	if err != nil {
		return err
	}
	if !enabled {
		return nil
	}
	rollupDate := rollupDateForReportingTime(event.CreatedAt, loc)
	rows := s.buildRollupFromEvents(event.AccountID, rollupDate, []model.ReportingEvent{*event})
	return s.rollupRepo.AdditiveUpsert(ctx, rows)
}

// ComputeDailyRollup computes rollup for a specific account and date.
func (s *ReportingRollupService) ComputeDailyRollup(ctx context.Context, accountID uint, date time.Time) error {
	loc, enabled, err := accountReportingLocation(ctx, reportingServiceDB(s.eventRepo, s.rollupRepo), accountID)
	if err != nil {
		return err
	}
	if !enabled {
		return nil
	}
	startUTC, endUTC, rollupDate := utcBoundariesForReportingDate(date, loc)

	// Get raw events for this time range
	events, err := s.eventRepo.FindByAccountIDAndTimeRange(ctx, accountID, startUTC, endUTC)
	if err != nil {
		applogger.L().Errorf("ComputeDailyRollup find events: %v", err)
		return err
	}

	// Build rollup rows using the backfill logic
	rollupRows := s.buildRollupFromEvents(accountID, rollupDate, events)

	// Delete existing rollups for this date and replace
	if err := s.rollupRepo.DeleteByAccountAndDate(ctx, accountID, rollupDate); err != nil {
		applogger.L().Errorf("ComputeDailyRollup delete old: %v", err)
		return err
	}

	if len(rollupRows) > 0 {
		if err := s.rollupRepo.BulkCreate(ctx, rollupRows); err != nil {
			applogger.L().Errorf("ComputeDailyRollup bulk insert: %v", err)
			return err
		}
	}

	return nil
}

// ComputeRollupForRange computes daily rollups for a date range.
func (s *ReportingRollupService) ComputeRollupForRange(ctx context.Context, accountID uint, since, until time.Time) error {
	for d := since; !d.After(until); d = d.AddDate(0, 0, 1) {
		if err := s.ComputeDailyRollup(ctx, accountID, d); err != nil {
			applogger.L().Errorf("ComputeRollupForRange date=%s: %v", d.Format("2006-01-02"), err)
		}
	}
	return nil
}

// GetSummaryMetrics retrieves aggregated metrics for a given time range and dimensions.
// Returns metrics matching Chatwoot's summary report API response format.
func (s *ReportingRollupService) GetSummaryMetrics(ctx context.Context, accountID uint, since, until time.Time, dimensionType model.DimensionType, dimensionID uint) (map[string]float64, error) {
	rollups, err := s.rollupRepo.FindByAccountAndDateRangeWithDimension(ctx, accountID, since, until, dimensionType, dimensionID)
	if err != nil {
		return nil, err
	}

	metrics := make(map[string]float64)
	for _, rollup := range rollups {
		metricKey := string(rollup.Metric)

		def := GetReportMetricDefinition(metricKey)
		if def == nil {
			// Unknown metric, still record
			metrics[metricKey] = float64(rollup.Count)
			continue
		}

		switch def.AggregateType {
		case "count":
			metrics[metricKey] = float64(rollup.Count)
		case "average":
			if rollup.Count > 0 {
				metrics[metricKey] = rollup.SumValue / float64(rollup.Count)
			}
		case "average_biz":
			if rollup.Count > 0 {
				metrics[metricKey] = rollup.SumValueBusinessHours / float64(rollup.Count)
			}
		case "sum":
			metrics[metricKey] = rollup.SumValue
		}
	}

	return metrics, nil
}

func (s *ReportingRollupService) buildRollupFromEvents(accountID uint, date time.Time, events []model.ReportingEvent) []model.ReportingEventsRollup {
	return rollupRowsFromAggregates(accountID, date, aggregateReportingEvents(accountID, events))
}