feat(reports): add analytics timeseries rollups

This commit is contained in:
2026-06-05 21:14:45 +08:00
parent a48b9cda8e
commit 43aa2f81eb
11 changed files with 494 additions and 32 deletions
+1
View File
@@ -604,6 +604,7 @@ func Bootstrap(env string) (*App, error) {
// Analytics services (P11 — Reports/Analytics)
analyticsService := service.NewAnalyticsService(reportingEventRepo, reportingEventsRollupRepo)
analyticsService.SetWorkerPool(workerPool)
summaryReportService := service.NewSummaryReportService(reportingEventsRollupRepo)
dashboardAppService := service.NewDashboardAppService(dashboardAppRepo)
platformAppService := service.NewPlatformAppService(platformAppRepo, accessTokenRepo, permissibleRepo)
@@ -2,6 +2,7 @@ package v1
import (
"net/http"
"strconv"
"time"
"github.com/gin-gonic/gin"
@@ -21,6 +22,41 @@ func NewAnalyticsHandler(svc *service.AnalyticsService) *AnalyticsHandler {
return &AnalyticsHandler{svc: svc}
}
// Index returns metric time-series for the overview charts.
// GET /api/v2/accounts/:account_id/reports?metric=...&since=...&until=...
func (h *AnalyticsHandler) Index(c *gin.Context) {
accountID, ok := parseAccountID(c)
if !ok {
return
}
since, until, ok := parseDateRange(c)
if !ok {
return
}
metric := c.Query("metric")
if metric == "" {
response.AbortWithStatusError(c, http.StatusUnprocessableEntity, response.ErrBadRequest, "metric parameter is required")
return
}
id := uint(0)
if rawID := c.Query("id"); rawID != "" {
parsed, err := strconv.ParseUint(rawID, 10, 64)
if err != nil {
response.AbortWithStatusError(c, http.StatusBadRequest, response.ErrBadRequest, "invalid id")
return
}
id = uint(parsed)
}
businessHours := c.Query("business_hours") == "true" || c.Query("business_hours") == "1"
result, err := h.svc.GetTimeseries(c.Request.Context(), accountID, metric, since, until, c.DefaultQuery("type", "account"), id, c.Query("group_by"), businessHours)
if err != nil {
applogger.L().Errorf("Timeseries report: %v", err)
response.AbortWithStatusError(c, http.StatusInternalServerError, response.ErrInternal, "failed to generate report")
return
}
response.OK(c, result)
}
// parseAccountID extracts account_id from URL params.
func parseAccountID(c *gin.Context) (uint, bool) {
id := getAccountID(c)
@@ -37,6 +37,7 @@ func (s *AnalyticsHandlerTestSuite) SetupSuite() {
s.db = db
s.Require().NoError(db.AutoMigrate(
&model.Account{},
&model.Conversation{},
&model.ReportingEvent{},
&model.ReportingEventsRollup{},
))
@@ -54,6 +55,7 @@ func (s *AnalyticsHandlerTestSuite) SetupSuite() {
gin.SetMode(gin.TestMode)
r := gin.New()
accounts := r.Group("/api/v1/accounts/:account_id")
accounts.GET("/reports", s.handler.Index)
accounts.GET("/reports/summary", s.handler.Summary)
accounts.GET("/reports/agents", s.handler.AgentMetrics)
accounts.GET("/reports/inboxes", s.handler.InboxMetrics)
@@ -70,6 +72,8 @@ func (s *AnalyticsHandlerTestSuite) TearDownSuite() {
func (s *AnalyticsHandlerTestSuite) SetupTest() {
s.db.Exec("DELETE FROM reporting_events")
s.db.Exec("DELETE FROM reporting_events_rollups")
s.db.Exec("DELETE FROM conversations")
}
// ========== parseAccountID / parseDateRange edge cases ==========
@@ -149,6 +153,23 @@ func (s *AnalyticsHandlerTestSuite) TestSummary_WithData() {
s.Equal(http.StatusOK, w.Code)
}
func (s *AnalyticsHandlerTestSuite) TestIndex_TimeseriesWithData() {
conv := model.Conversation{AccountID: s.accountID, InboxID: 1, ContactID: 1, Status: string(model.ConversationStatusOpen), ChannelType: "web_widget", Channel: "web_widget"}
s.Require().NoError(s.db.Create(&conv).Error)
s.Require().NoError(s.db.Model(&conv).Updates(map[string]interface{}{"created_at": parseTime("2025-01-15T10:00:00Z")}).Error)
w := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet,
"/api/v1/accounts/1/reports?metric=conversations_count&since=2025-01-01T00:00:00Z&until=2025-02-01T00:00:00Z&type=account&group_by=day", nil)
s.router.ServeHTTP(w, req)
s.Equal(http.StatusOK, w.Code)
var body map[string]interface{}
s.NoError(json.Unmarshal(w.Body.Bytes(), &body))
data := body["data"].([]interface{})
s.Len(data, 1)
s.Equal(float64(1), data[0].(map[string]interface{})["value"])
}
// ========== AgentMetrics ==========
func (s *AnalyticsHandlerTestSuite) TestAgentMetrics_InvalidAccountID() {
@@ -251,4 +272,4 @@ func (s *AnalyticsHandlerTestSuite) TestNilService() {
func parseTime(s string) time.Time {
t, _ := time.Parse(time.RFC3339, s)
return t
}
}
@@ -64,6 +64,7 @@ func (r *ReportingEventsRollupRepo) AggregateSummary(ctx context.Context, accoun
// DeleteByAccountAndDate deletes all rollups for a specific account and date.
func (r *ReportingEventsRollupRepo) DeleteByAccountAndDate(ctx context.Context, accountID uint, date time.Time) error {
return r.db.WithContext(ctx).
Unscoped().
Where("account_id = ? AND date = ?", accountID, date).
Delete(&model.ReportingEventsRollup{}).Error
}
+2 -1
View File
@@ -1323,6 +1323,7 @@ func registerV1Routes(g *gin.RouterGroup, h *Handlers) {
// Reference: Chatwoot reports_controller.rb + live_reports_controller.rb + summary_reports_controller.rb
reports := accountScoped.Group("/reports")
{
reports.GET("", h.Analytics.Index)
reports.GET("/summary", h.Analytics.Summary)
reports.GET("/bot_summary", h.Analytics.BotSummary)
reports.GET("/agents", h.Analytics.AgentMetrics)
@@ -1714,7 +1715,7 @@ func registerV2Routes(g *gin.RouterGroup, h *Handlers) {
reports := accountScoped.Group("/reports")
{
reports.GET("", h.Analytics.Summary)
reports.GET("", h.Analytics.Index)
reports.GET("/summary", h.Analytics.Summary)
reports.GET("/bot_summary", h.Analytics.BotSummary)
reports.GET("/agents", h.Analytics.AgentMetrics)
+43
View File
@@ -7,6 +7,7 @@ import (
"github.com/gochat/gochat/internal/model"
"github.com/gochat/gochat/internal/repository"
"github.com/gochat/gochat/internal/worker"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"gorm.io/driver/sqlite"
@@ -32,6 +33,7 @@ func setupAnalyticsP513Test(t *testing.T) (*gorm.DB, *AnalyticsService, *model.A
&model.AgentBotInbox{},
&model.Tag{},
&model.ConversationLabel{},
&model.BackgroundJob{},
))
t.Cleanup(func() {
sqlDB, _ := db.DB()
@@ -147,3 +149,44 @@ func TestAnalyticsReportsUsePersistedConversationMessageAndEventRows(t *testing.
matrixMap := matrix.(map[string]interface{})
assert.Equal(t, [][]int64{{1}}, matrixMap["matrix"])
}
func TestAnalyticsTimeseriesAndRollupWorker(t *testing.T) {
db, svc, account, inbox, contact, user, _ := setupAnalyticsP513Test(t)
since := time.Date(2026, 6, 2, 0, 0, 0, 0, time.UTC)
until := since.Add(48 * time.Hour)
conv := &model.Conversation{AccountID: account.ID, InboxID: inbox.ID, ContactID: contact.ID, AssigneeID: &user.ID, Status: string(model.ConversationStatusOpen), ChannelType: inbox.ChannelType, Channel: inbox.ChannelType}
require.NoError(t, db.Create(conv).Error)
require.NoError(t, db.Model(conv).Updates(map[string]interface{}{"created_at": since.Add(2 * time.Hour)}).Error)
require.NoError(t, db.Create(&model.ReportingEvent{Base: model.Base{CreatedAt: since.Add(3 * time.Hour)}, AccountID: account.ID, Name: model.MetricNameFirstResponse, Value: 120, ConversationID: &conv.ID, InboxID: &inbox.ID, UserID: &user.ID, EventStartTime: since, EventEndTime: since.Add(2 * time.Minute)}).Error)
points, err := svc.GetTimeseries(context.Background(), account.ID, "conversations_count", since, until, "account", 0, "day", false)
require.NoError(t, err)
require.Len(t, points, 1)
assert.Equal(t, float64(1), points[0].Value)
assert.Equal(t, since.Unix(), points[0].Timestamp)
avgPoints, err := svc.GetTimeseries(context.Background(), account.ID, "avg_first_response_time", since, until, "agent", user.ID, "day", false)
require.NoError(t, err)
require.Len(t, avgPoints, 1)
assert.Equal(t, 120.0, avgPoints[0].Value)
assert.Equal(t, int64(1), avgPoints[0].Count)
wp := worker.NewWorkerPool(db)
svc.SetWorkerPool(wp)
job, err := EnqueueReportingRollupDay(context.Background(), wp, account.ID, since)
require.NoError(t, err)
require.NotNil(t, job)
again, err := EnqueueReportingRollupDay(context.Background(), wp, account.ID, since)
require.NoError(t, err)
assert.Equal(t, job.ID, again.ID)
processed, err := wp.ProcessOne(context.Background())
require.NoError(t, err)
assert.True(t, processed)
processed, err = wp.ProcessOne(context.Background())
require.NoError(t, err)
assert.False(t, processed)
var rollupCount int64
require.NoError(t, db.Model(&model.ReportingEventsRollup{}).Where("account_id = ? AND date = ?", account.ID, since).Count(&rollupCount).Error)
assert.Greater(t, rollupCount, int64(0))
}
+269
View File
@@ -3,7 +3,9 @@ package service
import (
"context"
"errors"
"fmt"
"sort"
"strconv"
"strings"
"time"
@@ -11,6 +13,12 @@ import (
"gorm.io/gorm"
)
type AnalyticsTimeseriesPoint struct {
Value float64 `json:"value"`
Timestamp int64 `json:"timestamp"`
Count int64 `json:"count,omitempty"`
}
func (s *AnalyticsService) analyticsDB() (*gorm.DB, error) {
if s == nil || s.db == nil {
return nil, errors.New("analytics database is required")
@@ -18,6 +26,64 @@ func (s *AnalyticsService) analyticsDB() (*gorm.DB, error) {
return s.db, nil
}
func (s *AnalyticsService) EnsureRollupsForRange(ctx context.Context, accountID uint, since, until time.Time) error {
if s == nil || s.rollups == nil || s.db == nil || since.IsZero() || until.IsZero() {
return nil
}
start := dayStart(since)
end := dayStart(until)
if until.After(end) {
end = end.AddDate(0, 0, 1)
}
for d := start; d.Before(end); d = d.AddDate(0, 0, 1) {
var count int64
if err := s.db.WithContext(ctx).Model(&model.ReportingEventsRollup{}).Where("account_id = ? AND date = ?", accountID, d).Count(&count).Error; err != nil {
return err
}
if count > 0 {
continue
}
if err := s.rollups.ComputeDailyRollup(ctx, accountID, d); err != nil {
return err
}
}
return nil
}
func (s *AnalyticsService) GetTimeseries(ctx context.Context, accountID uint, metric string, since, until time.Time, reportType string, id uint, groupBy string, businessHours bool) ([]AnalyticsTimeseriesPoint, error) {
if strings.TrimSpace(groupBy) == "" {
groupBy = "day"
}
if reportType == "" {
reportType = "account"
}
if err := s.EnsureRollupsForRange(ctx, accountID, since, until); err != nil {
return nil, err
}
switch metric {
case "conversations_count":
return s.conversationCountTimeseries(ctx, accountID, since, until, reportType, id, groupBy, false)
case "resolutions_count":
return s.conversationCountTimeseries(ctx, accountID, since, until, reportType, id, groupBy, true)
case "incoming_messages_count":
return s.messageCountTimeseries(ctx, accountID, since, until, reportType, id, groupBy, model.MessageTypeIncoming)
case "outgoing_messages_count":
return s.messageCountTimeseries(ctx, accountID, since, until, reportType, id, groupBy, model.MessageTypeOutgoing)
case "avg_first_response_time":
return s.eventTimeseries(ctx, accountID, since, until, reportType, id, groupBy, []string{model.MetricNameFirstResponse}, true, businessHours)
case "avg_resolution_time":
return s.eventTimeseries(ctx, accountID, since, until, reportType, id, groupBy, []string{"conversation_resolved", model.MetricNameResolutionTime}, true, businessHours)
case "reply_time":
return s.eventTimeseries(ctx, accountID, since, until, reportType, id, groupBy, []string{model.MetricNameReplyTime}, true, businessHours)
case "bot_resolutions_count":
return s.eventTimeseries(ctx, accountID, since, until, reportType, id, groupBy, []string{"conversation_bot_resolved", model.MetricNameBotResolutionsCount}, false, businessHours)
case "bot_handoffs_count":
return s.eventTimeseries(ctx, accountID, since, until, reportType, id, groupBy, []string{"conversation_bot_handoff", model.MetricNameBotHandoffsCount}, false, businessHours)
default:
return nil, fmt.Errorf("unsupported report metric %q", metric)
}
}
func (s *AnalyticsService) liveConversationMetrics(ctx context.Context, accountID uint, teamID uint) (*ConversationMetrics, error) {
db, err := s.analyticsDB()
if err != nil {
@@ -46,6 +112,209 @@ func (s *AnalyticsService) liveConversationMetrics(ctx context.Context, accountI
return &result, nil
}
func (s *AnalyticsService) conversationCountTimeseries(ctx context.Context, accountID uint, since, until time.Time, reportType string, id uint, groupBy string, resolved bool) ([]AnalyticsTimeseriesPoint, error) {
db, err := s.analyticsDB()
if err != nil {
return nil, err
}
var conversations []model.Conversation
q := db.WithContext(ctx).Model(&model.Conversation{}).Where("account_id = ?", accountID)
if resolved {
q = q.Where("resolved_at IS NOT NULL AND resolved_at >= ? AND resolved_at < ?", since, until)
} else {
q = q.Where("created_at >= ? AND created_at < ?", since, until)
}
q = applyConversationDimension(q, reportType, id)
if err := q.Find(&conversations).Error; err != nil {
return nil, err
}
buckets := map[time.Time]*AnalyticsTimeseriesPoint{}
for _, conversation := range conversations {
t := conversation.CreatedAt
if resolved && conversation.ResolvedAt != nil {
t = *conversation.ResolvedAt
}
bucket := bucketStart(t, groupBy)
point := ensureTimeseriesBucket(buckets, bucket)
point.Value++
}
return sortedTimeseries(buckets), nil
}
func (s *AnalyticsService) messageCountTimeseries(ctx context.Context, accountID uint, since, until time.Time, reportType string, id uint, groupBy string, messageType model.MessageType) ([]AnalyticsTimeseriesPoint, error) {
db, err := s.analyticsDB()
if err != nil {
return nil, err
}
var messages []model.Message
q := db.WithContext(ctx).Model(&model.Message{}).Where("messages.account_id = ? AND messages.message_type = ? AND messages.created_at >= ? AND messages.created_at < ?", accountID, string(messageType), since, until)
q = applyMessageDimension(q, reportType, id)
if err := q.Find(&messages).Error; err != nil {
return nil, err
}
buckets := map[time.Time]*AnalyticsTimeseriesPoint{}
for _, message := range messages {
point := ensureTimeseriesBucket(buckets, bucketStart(message.CreatedAt, groupBy))
point.Value++
}
return sortedTimeseries(buckets), nil
}
func (s *AnalyticsService) eventTimeseries(ctx context.Context, accountID uint, since, until time.Time, reportType string, id uint, groupBy string, names []string, average bool, businessHours bool) ([]AnalyticsTimeseriesPoint, error) {
db, err := s.analyticsDB()
if err != nil {
return nil, err
}
var events []model.ReportingEvent
q := db.WithContext(ctx).Model(&model.ReportingEvent{}).Where("account_id = ? AND name IN ? AND created_at >= ? AND created_at < ?", accountID, names, since, until)
q = applyEventDimension(q, reportType, id)
if err := q.Find(&events).Error; err != nil {
return nil, err
}
buckets := map[time.Time]*AnalyticsTimeseriesPoint{}
for _, event := range events {
point := ensureTimeseriesBucket(buckets, bucketStart(event.CreatedAt, groupBy))
value := event.Value
if businessHours {
value = event.ValueInBusinessHours
}
if average {
point.Value += value
point.Count++
} else {
point.Value++
}
}
if average {
for _, point := range buckets {
if point.Count > 0 {
point.Value = point.Value / float64(point.Count)
}
}
}
return sortedTimeseries(buckets), nil
}
func applyConversationDimension(query *gorm.DB, reportType string, id uint) *gorm.DB {
switch reportType {
case "inbox":
if id > 0 {
query = query.Where("inbox_id = ?", id)
}
case "agent":
if id > 0 {
query = query.Where("assignee_id = ?", id)
}
case "team":
if id > 0 {
query = query.Where("team_id = ?", id)
}
case "label":
if id > 0 {
query = query.Joins("INNER JOIN conversation_labels ON conversation_labels.conversation_id = conversations.id").Where("conversation_labels.tag_id = ?", id)
}
}
return query
}
func applyMessageDimension(query *gorm.DB, reportType string, id uint) *gorm.DB {
switch reportType {
case "inbox":
if id > 0 {
query = query.Where("messages.inbox_id = ?", id)
}
case "agent":
if id > 0 {
query = query.Where("messages.sender_type = ? AND messages.sender_id = ?", "User", id)
}
case "team":
if id > 0 {
query = query.Joins("INNER JOIN conversations ON conversations.id = messages.conversation_id").Where("conversations.team_id = ?", id)
}
case "label":
if id > 0 {
query = query.Joins("INNER JOIN conversation_labels ON conversation_labels.conversation_id = messages.conversation_id").Where("conversation_labels.tag_id = ?", id)
}
}
return query
}
func applyEventDimension(query *gorm.DB, reportType string, id uint) *gorm.DB {
switch reportType {
case "inbox":
if id > 0 {
query = query.Where("inbox_id = ?", id)
}
case "agent":
if id > 0 {
query = query.Where("user_id = ?", id)
}
case "team":
if id > 0 {
query = query.Joins("INNER JOIN conversations ON conversations.id = reporting_events.conversation_id").Where("conversations.team_id = ?", id)
}
case "label":
if id > 0 {
query = query.Joins("INNER JOIN conversation_labels ON conversation_labels.conversation_id = reporting_events.conversation_id").Where("conversation_labels.tag_id = ?", id)
}
}
return query
}
func dayStart(t time.Time) time.Time {
return time.Date(t.Year(), t.Month(), t.Day(), 0, 0, 0, 0, time.UTC)
}
func bucketStart(t time.Time, groupBy string) time.Time {
t = t.UTC()
switch groupBy {
case "hour":
return time.Date(t.Year(), t.Month(), t.Day(), t.Hour(), 0, 0, 0, time.UTC)
case "week":
start := dayStart(t)
return start.AddDate(0, 0, -int(start.Weekday()))
case "month":
return time.Date(t.Year(), t.Month(), 1, 0, 0, 0, 0, time.UTC)
case "year":
return time.Date(t.Year(), 1, 1, 0, 0, 0, 0, time.UTC)
default:
return dayStart(t)
}
}
func ensureTimeseriesBucket(buckets map[time.Time]*AnalyticsTimeseriesPoint, bucket time.Time) *AnalyticsTimeseriesPoint {
point := buckets[bucket]
if point == nil {
point = &AnalyticsTimeseriesPoint{Timestamp: bucket.Unix()}
buckets[bucket] = point
}
return point
}
func sortedTimeseries(buckets map[time.Time]*AnalyticsTimeseriesPoint) []AnalyticsTimeseriesPoint {
keys := make([]time.Time, 0, len(buckets))
for key := range buckets {
keys = append(keys, key)
}
sort.Slice(keys, func(i, j int) bool { return keys[i].Before(keys[j]) })
result := make([]AnalyticsTimeseriesPoint, 0, len(keys))
for _, key := range keys {
result = append(result, *buckets[key])
}
return result
}
func parseReportDimensionID(raw string) uint {
if strings.TrimSpace(raw) == "" {
return 0
}
parsed, err := strconv.ParseUint(raw, 10, 64)
if err != nil {
return 0
}
return uint(parsed)
}
func (s *AnalyticsService) groupedLiveConversationMetrics(ctx context.Context, accountID uint, groupBy string) ([]map[string]interface{}, error) {
if groupBy != "team_id" && groupBy != "assignee_id" {
return nil, errors.New("invalid group_by")
+18
View File
@@ -6,6 +6,7 @@ import (
"github.com/gochat/gochat/internal/model"
"github.com/gochat/gochat/internal/repository"
"github.com/gochat/gochat/internal/worker"
applogger "github.com/gochat/gochat/pkg/logger"
"gorm.io/gorm"
)
@@ -16,6 +17,8 @@ type AnalyticsService struct {
eventRepo *repository.ReportingEventRepo
rollupRepo *repository.ReportingEventsRollupRepo
db *gorm.DB
rollups *ReportingRollupService
worker *worker.WorkerPool
}
func NewAnalyticsService(
@@ -33,9 +36,15 @@ func NewAnalyticsService(
eventRepo: eventRepo,
rollupRepo: rollupRepo,
db: db,
rollups: NewReportingRollupService(eventRepo, rollupRepo),
}
}
func (s *AnalyticsService) SetWorkerPool(wp *worker.WorkerPool) {
s.worker = wp
RegisterReportingRollupJobs(wp, s)
}
// --- Response DTOs ---
// MetricSummary holds aggregated metric data.
@@ -84,6 +93,9 @@ type GroupedConversationMetric struct {
// GetSummary returns account-level aggregated metrics for a date range.
func (s *AnalyticsService) GetSummary(ctx context.Context, accountID uint, since, until time.Time) (*SummaryResponse, error) {
if err := s.EnsureRollupsForRange(ctx, accountID, since, until); err != nil {
return nil, err
}
rollups, err := s.rollupRepo.FindByAccountAndDateRange(ctx, accountID, since, until)
if err != nil {
applogger.L().Errorf("GetSummary rollup query: %v", err)
@@ -143,6 +155,9 @@ func (s *AnalyticsService) GetTeamMetrics(ctx context.Context, accountID uint, s
}
func (s *AnalyticsService) getDimensionMetrics(ctx context.Context, accountID uint, dimType model.DimensionType, since, until time.Time) ([]DimensionMetrics, error) {
if err := s.EnsureRollupsForRange(ctx, accountID, since, until); err != nil {
return nil, err
}
rollups, err := s.rollupRepo.AggregateSummary(ctx, accountID, dimType, since, until)
if err != nil {
applogger.L().Errorf("getDimensionMetrics query: %v", err)
@@ -192,6 +207,9 @@ func (s *AnalyticsService) getDimensionMetrics(ctx context.Context, accountID ui
// GetConversationTraffic returns daily conversation count time-series.
func (s *AnalyticsService) GetConversationTraffic(ctx context.Context, accountID uint, since, until time.Time) ([]ConversationTrafficPoint, error) {
if err := s.EnsureRollupsForRange(ctx, accountID, since, until); err != nil {
return nil, err
}
rollups, err := s.rollupRepo.FindByMetric(ctx, accountID, model.MetricResolutionsCount, since, until)
if err != nil {
applogger.L().Errorf("GetConversationTraffic query: %v", err)
@@ -0,0 +1,62 @@
package service
import (
"context"
"encoding/json"
"fmt"
"sync"
"time"
"github.com/gochat/gochat/internal/model"
"github.com/gochat/gochat/internal/worker"
)
const TaskTypeReportingRollupDay = "reporting:rollup_day"
type reportingRollupDayJob struct {
AccountID uint `json:"account_id"`
Date string `json:"date"`
}
var reportingRollupRegistrations sync.Map
func RegisterReportingRollupJobs(wp *worker.WorkerPool, svc *AnalyticsService) {
if wp == nil || svc == nil {
return
}
if _, loaded := reportingRollupRegistrations.LoadOrStore(wp, struct{}{}); loaded {
return
}
wp.Register(TaskTypeReportingRollupDay, svc.performReportingRollupDayJob)
}
func EnqueueReportingRollupDay(ctx context.Context, wp *worker.WorkerPool, accountID uint, date time.Time) (*model.BackgroundJob, error) {
if wp == nil || accountID == 0 {
return nil, nil
}
day := time.Date(date.Year(), date.Month(), date.Day(), 0, 0, 0, 0, time.UTC)
dateStr := day.Format(time.DateOnly)
return wp.Enqueue(ctx, TaskTypeReportingRollupDay, reportingRollupDayJob{AccountID: accountID, Date: dateStr},
worker.WithQueue("low"),
worker.WithMaxAttempts(3),
worker.WithIdempotencyKey(fmt.Sprintf("reporting:rollup_day:%d:%s", accountID, dateStr)),
)
}
func (s *AnalyticsService) performReportingRollupDayJob(ctx context.Context, job *model.BackgroundJob) error {
var payload reportingRollupDayJob
if err := json.Unmarshal(job.Payload, &payload); err != nil {
return fmt.Errorf("unmarshal reporting rollup job: %w", err)
}
if payload.AccountID == 0 || payload.Date == "" {
return fmt.Errorf("invalid reporting rollup payload: %#v", payload)
}
date, err := time.Parse(time.DateOnly, payload.Date)
if err != nil {
return fmt.Errorf("parse reporting rollup date: %w", err)
}
if s == nil || s.rollups == nil {
return fmt.Errorf("reporting rollup service is required")
}
return s.rollups.ComputeDailyRollup(ctx, payload.AccountID, date)
}