package app import ( "context" "encoding/json" "fmt" "sort" "time" "github.com/geo-platform/tenant-api/internal/shared/response" "go.uber.org/zap" ) const ( monitoringProjectionQueueIOTimeout = 5 * time.Second monitoringProjectionFallbackTimeout = 2 * time.Minute monitoringProjectionQueueMessageVer = "v1" ) type monitoringProjectionRebuildEnvelope struct { Version string `json:"version"` TenantID int64 `json:"tenant_id"` BrandID int64 `json:"brand_id"` BusinessDate string `json:"business_date"` Source string `json:"source"` } func encodeMonitoringProjectionRebuildEnvelope(tenantID, brandID int64, businessDay time.Time, source string) ([]byte, error) { businessDay = monitoringCanonicalBusinessDay(businessDay) return json.Marshal(monitoringProjectionRebuildEnvelope{ Version: monitoringProjectionQueueMessageVer, TenantID: tenantID, BrandID: brandID, BusinessDate: businessDay.Format("2006-01-02"), Source: source, }) } func decodeMonitoringProjectionRebuildEnvelope(payload []byte) (*monitoringProjectionRebuildEnvelope, error) { if len(payload) == 0 { return nil, fmt.Errorf("monitoring projection rebuild payload is empty") } var envelope monitoringProjectionRebuildEnvelope if err := json.Unmarshal(payload, &envelope); err != nil { return nil, err } if envelope.TenantID <= 0 { return nil, fmt.Errorf("monitoring projection rebuild payload missing tenant_id") } if envelope.BrandID <= 0 { return nil, fmt.Errorf("monitoring projection rebuild payload missing brand_id") } if _, err := parseMonitoringBusinessDateInLocation(envelope.BusinessDate, monitoringBusinessLocation()); err != nil { return nil, fmt.Errorf("monitoring projection rebuild payload has invalid business_date") } return &envelope, nil } func (s *MonitoringCallbackService) publishProjectionRebuildEnvelope(ctx context.Context, payload []byte) error { if s == nil || s.rabbitMQ == nil { return response.ErrServiceUnavailable(50341, "monitoring_projection_queue_unavailable", "rabbitmq client is not configured") } if err := s.rabbitMQ.PublishMonitorProjectionRebuild(ctx, payload); err != nil { return response.ErrServiceUnavailable(50341, "monitoring_projection_queue_unavailable", err.Error()) } return nil } func (s *MonitoringCallbackService) dispatchMonitoringProjectionRebuilds(tenantID int64, businessDay time.Time, brandIDs map[int64]struct{}, source string) { if s == nil || len(brandIDs) == 0 { return } sortedBrandIDs := sortedMonitoringProjectionBrandIDs(brandIDs) failedBrandIDs := make([]int64, 0) for _, brandID := range sortedBrandIDs { payload, err := encodeMonitoringProjectionRebuildEnvelope(tenantID, brandID, businessDay, source) if err != nil { failedBrandIDs = append(failedBrandIDs, brandID) if s.logger != nil { s.logger.Warn("monitoring projection rebuild payload encode failed", zap.Int64("tenant_id", tenantID), zap.Int64("brand_id", brandID), zap.Error(err)) } continue } ctx, cancel := context.WithTimeout(context.Background(), monitoringProjectionQueueIOTimeout) err = s.publishProjectionRebuildEnvelope(ctx, payload) cancel() if err != nil { failedBrandIDs = append(failedBrandIDs, brandID) if s.logger != nil { s.logger.Warn("monitoring projection rebuild enqueue failed", zap.Int64("tenant_id", tenantID), zap.Int64("brand_id", brandID), zap.Error(err)) } } } if len(failedBrandIDs) == 0 { return } ctx, cancel := context.WithTimeout(context.Background(), monitoringProjectionFallbackTimeout) defer cancel() if err := s.rebuildBrandDailyProjectionBatch(ctx, tenantID, businessDay, failedBrandIDs); err != nil { if s.logger != nil { s.logger.Warn("monitoring projection rebuild fallback failed", zap.Int64("tenant_id", tenantID), zap.Int("brand_count", len(failedBrandIDs)), zap.Error(err), ) } return } if s.logger != nil { s.logger.Info("monitoring projection rebuild fallback completed", zap.Int64("tenant_id", tenantID), zap.Int("brand_count", len(failedBrandIDs)), ) } } func (s *MonitoringCallbackService) ProcessProjectionRebuildEnvelope(ctx context.Context, envelope monitoringProjectionRebuildEnvelope) error { businessDay, err := parseMonitoringBusinessDateInLocation(envelope.BusinessDate, monitoringBusinessLocation()) if err != nil { return response.ErrBadRequest(40041, "invalid_business_date", "projection rebuild business_date is invalid") } return s.rebuildBrandDailyProjectionBatch(ctx, envelope.TenantID, businessDay, []int64{envelope.BrandID}) } func (s *MonitoringCallbackService) rebuildBrandDailyProjectionBatch(ctx context.Context, tenantID int64, businessDay time.Time, brandIDs []int64) error { if s == nil || s.monitoringPool == nil || len(brandIDs) == 0 { return nil } businessDay = monitoringCanonicalBusinessDay(businessDay) tx, err := s.monitoringPool.Begin(ctx) if err != nil { return response.ErrInternal(50041, "begin_failed", "failed to start monitoring projection rebuild") } defer tx.Rollback(ctx) for _, brandID := range uniqueSortedMonitoringProjectionBrandIDs(brandIDs) { if err := s.rebuildBrandDailyProjection(ctx, tx, tenantID, brandID, businessDay); err != nil { return err } } if err := tx.Commit(ctx); err != nil { return response.ErrInternal(50041, "commit_failed", "failed to commit monitoring projection rebuild") } return nil } func sortedMonitoringProjectionBrandIDs(items map[int64]struct{}) []int64 { result := make([]int64, 0, len(items)) for brandID := range items { result = append(result, brandID) } sort.Slice(result, func(i, j int) bool { return result[i] < result[j] }) return result } func uniqueSortedMonitoringProjectionBrandIDs(items []int64) []int64 { if len(items) == 0 { return nil } seen := make(map[int64]struct{}, len(items)) result := make([]int64, 0, len(items)) for _, brandID := range items { if brandID <= 0 { continue } if _, ok := seen[brandID]; ok { continue } seen[brandID] = struct{}{} result = append(result, brandID) } sort.Slice(result, func(i, j int) bool { return result[i] < result[j] }) return result }