feat(reports): add live rollup upserts

This commit is contained in:
2026-06-07 00:11:28 +08:00
parent d75c5f4287
commit c9a86793ce
7 changed files with 125 additions and 13 deletions
@@ -6,6 +6,7 @@ import (
"github.com/gochat/gochat/internal/model"
"gorm.io/gorm"
"gorm.io/gorm/clause"
)
// ReportingEventsRollupRepo provides data access for ReportingEventsRollup.
@@ -77,6 +78,29 @@ func (r *ReportingEventsRollupRepo) BulkCreate(ctx context.Context, rollups []mo
return r.db.WithContext(ctx).CreateInBatches(rollups, 100).Error
}
// AdditiveUpsert inserts rollups or adds incoming values to existing rows.
// Mirrors Chatwoot ReportingEvents::RollupService upsert_all on_duplicate SQL.
func (r *ReportingEventsRollupRepo) AdditiveUpsert(ctx context.Context, rollups []model.ReportingEventsRollup) error {
if len(rollups) == 0 {
return nil
}
return r.db.WithContext(ctx).Clauses(clause.OnConflict{
Columns: []clause.Column{
{Name: "account_id"},
{Name: "date"},
{Name: "dimension_type"},
{Name: "dimension_id"},
{Name: "metric"},
},
DoUpdates: clause.Assignments(map[string]interface{}{
"count": gorm.Expr("reporting_events_rollups.count + excluded.count"),
"sum_value": gorm.Expr("reporting_events_rollups.sum_value + excluded.sum_value"),
"sum_value_business_hours": gorm.Expr("reporting_events_rollups.sum_value_business_hours + excluded.sum_value_business_hours"),
"updated_at": gorm.Expr("excluded.updated_at"),
}),
}).CreateInBatches(rollups, 100).Error
}
// FindByAccountAndDateRangeWithDimension retrieves rollups with dimension filters.
func (r *ReportingEventsRollupRepo) FindByAccountAndDateRangeWithDimension(ctx context.Context, accountID uint, since, until time.Time, dimType model.DimensionType, dimID uint) ([]model.ReportingEventsRollup, error) {
var rollups []model.ReportingEventsRollup
+5
View File
@@ -249,6 +249,11 @@ func (s *AnalyticsService) RecordEvent(ctx context.Context, event *model.Reporti
applogger.L().Errorf("RecordEvent: %v", err)
return err
}
if s.rollups != nil {
if err := s.rollups.RollupEvent(ctx, event); err != nil {
applogger.L().Errorf("RecordEvent rollup: %v", err)
}
}
return nil
}
@@ -67,13 +67,6 @@ func ExpandEventToRollupMetrics(rawMetric model.RollupMetric, count int64, sumVa
SumBizHours: 0,
}
default:
// Unknown metric — still record it as a direct rollup
result[rawMetric] = RollupMetricData{
Count: count,
SumValue: sumValue,
SumBizHours: sumBizHours,
}
}
return result
@@ -47,6 +47,11 @@ func utcBoundariesForReportingDate(date time.Time, loc *time.Location) (time.Tim
return startUTC, endUTC, rollupDate
}
func rollupDateForReportingTime(value time.Time, loc *time.Location) time.Time {
localTime := value.In(loc)
return time.Date(localTime.Year(), localTime.Month(), localTime.Day(), 0, 0, 0, 0, time.UTC)
}
func dimensionIDForEvent(accountID uint, event model.ReportingEvent, dim DimensionSpec) (uint, bool) {
switch dim.GroupColumn {
case "":
@@ -26,6 +26,24 @@ func NewReportingRollupService(eventRepo *repository.ReportingEventRepo, rollupR
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)
@@ -108,6 +108,70 @@ func TestReportingRollupUsesReportingTimezoneCreatedAtBoundaries(t *testing.T) {
assert.Equal(t, 0.0, resolutionsCount.SumValue)
}
func TestAnalyticsRecordEventAppliesAdditiveLiveRollup(t *testing.T) {
db, _, _ := setupReportingRollupTZTest(t)
account := createReportingAccount(t, db, "America/New_York")
user, inbox, conversation := createReportingActorRows(t, db, account.ID)
rollupDate := time.Date(2026, 2, 11, 0, 0, 0, 0, time.UTC)
require.NoError(t, db.Create(&model.ReportingEventsRollup{
AccountID: account.ID,
Date: rollupDate,
DimensionType: model.DimensionAccount,
DimensionID: account.ID,
Metric: model.MetricResolutionTime,
Count: 7,
SumValue: 700,
SumValueBusinessHours: 350,
}).Error)
svc := NewAnalyticsService(repository.NewReportingEventRepo(db), repository.NewReportingEventsRollupRepo(db))
createdAt := time.Date(2026, 2, 12, 4, 0, 0, 0, time.UTC)
event := &model.ReportingEvent{
Base: model.Base{CreatedAt: createdAt},
AccountID: account.ID,
Name: "conversation_resolved",
Value: 100,
ValueInBusinessHours: 50,
UserID: &user.ID,
InboxID: &inbox.ID,
ConversationID: &conversation.ID,
EventStartTime: createdAt.Add(-time.Hour),
EventEndTime: createdAt,
}
require.NoError(t, svc.RecordEvent(context.Background(), event))
var rawCount int64
require.NoError(t, db.Model(&model.ReportingEvent{}).Where("account_id = ? AND name = ?", account.ID, "conversation_resolved").Count(&rawCount).Error)
assert.Equal(t, int64(1), rawCount)
var resolutionTime model.ReportingEventsRollup
require.NoError(t, db.Where("account_id = ? AND date = ? AND dimension_type = ? AND dimension_id = ? AND metric = ?", account.ID, rollupDate, model.DimensionAccount, account.ID, model.MetricResolutionTime).First(&resolutionTime).Error)
assert.Equal(t, int64(8), resolutionTime.Count)
assert.Equal(t, 800.0, resolutionTime.SumValue)
assert.Equal(t, 400.0, resolutionTime.SumValueBusinessHours)
var resolutionsCount model.ReportingEventsRollup
require.NoError(t, db.Where("account_id = ? AND date = ? AND dimension_type = ? AND dimension_id = ? AND metric = ?", account.ID, rollupDate, model.DimensionAccount, account.ID, model.MetricResolutionsCount).First(&resolutionsCount).Error)
assert.Equal(t, int64(1), resolutionsCount.Count)
assert.Equal(t, 0.0, resolutionsCount.SumValue)
var agentRows int64
require.NoError(t, db.Model(&model.ReportingEventsRollup{}).Where("account_id = ? AND date = ? AND dimension_type = ? AND dimension_id = ?", account.ID, rollupDate, model.DimensionAgent, user.ID).Count(&agentRows).Error)
assert.Equal(t, int64(2), agentRows)
}
func TestReportingRollupIgnoresUnknownEventMetrics(t *testing.T) {
db, rollups, _ := setupReportingRollupTZTest(t)
account := createReportingAccount(t, db, "America/New_York")
createReportingEvent(t, db, account.ID, "conversation_opened", time.Date(2026, 2, 11, 15, 0, 0, 0, time.UTC), nil, nil, nil)
require.NoError(t, rollups.ComputeDailyRollup(context.Background(), account.ID, time.Date(2026, 2, 11, 0, 0, 0, 0, time.UTC)))
var count int64
require.NoError(t, db.Model(&model.ReportingEventsRollup{}).Where("account_id = ?", account.ID).Count(&count).Error)
assert.Equal(t, int64(0), count)
}
func TestReportingRollupSkipsNilAgentAndInboxDimensions(t *testing.T) {
db, rollups, _ := setupReportingRollupTZTest(t)
account := createReportingAccount(t, db, "America/New_York")