From 927319807bc33ed869cd13059a080d120c8d7b0c Mon Sep 17 00:00:00 2001 From: Amirhossein Khalili Date: Fri, 21 Aug 2026 07:49:04 +0800 Subject: [PATCH] fix: unblock first growth report generation --- internal/growth/analytics_test.go | 15 +++++++++++++ internal/growth/report.go | 9 ++++++++ internal/growthreport/coordinator.go | 4 ++-- internal/growthreport/coordinator_test.go | 9 ++++++++ internal/growthreport/orchestrator.go | 21 +++++++++++++------ internal/growthreport/repository.go | 16 ++++++++++---- .../repository_integration_test.go | 7 +++++-- 7 files changed, 67 insertions(+), 14 deletions(-) diff --git a/internal/growth/analytics_test.go b/internal/growth/analytics_test.go index 3f8f074..22eb535 100644 --- a/internal/growth/analytics_test.go +++ b/internal/growth/analytics_test.go @@ -1,6 +1,8 @@ package growth import ( + "bytes" + "encoding/json" "testing" "time" ) @@ -194,6 +196,19 @@ func TestBuildDeterministicReportLimitsAndOrdersOpportunities(t *testing.T) { } } +func TestBuildDeterministicReportSerializesEmptyCollectionsAsArrays(t *testing.T) { + report := BuildDeterministicReport(ReportInput{ForecastStart: time.Date(2026, 7, 1, 0, 0, 0, 0, time.UTC), HorizonDays: 90}) + encoded, err := json.Marshal(report) + if err != nil { + t.Fatal(err) + } + for _, field := range []string{`"historicalWins":[]`, `"opportunities":[]`, `"risks":[]`, `"fallbackBriefs":[]`} { + if !bytes.Contains(encoded, []byte(field)) { + t.Fatalf("report must encode %s as an array: %s", field, encoded) + } + } +} + func TestFingerprintTreatsEquivalentForecastDatesEqually(t *testing.T) { tehran, err := time.LoadLocation("Asia/Tehran") if err != nil { diff --git a/internal/growth/report.go b/internal/growth/report.go index 793db53..c143c6c 100644 --- a/internal/growth/report.go +++ b/internal/growth/report.go @@ -64,6 +64,15 @@ func BuildDeterministicReport(input ReportInput) DeterministicReport { wins := historicalWins(input.History, catalog, input.ForecastStart) opportunities := buildOpportunities(wins, catalog, forecast, input.ForecastStart, 5) risks := DetectRisks(forecast, 3) + if wins == nil { + wins = []EventEffect{} + } + if opportunities == nil { + opportunities = []Opportunity{} + } + if risks == nil { + risks = []Risk{} + } bundle := EvidenceBundle{Opportunities: opportunities, Risks: risks} report := DeterministicReport{Forecast: forecast, HistoricalWins: wins, Opportunities: opportunities, Risks: risks, FallbackBriefs: BuildFallback(bundle)} if input.Peer != nil { diff --git a/internal/growthreport/coordinator.go b/internal/growthreport/coordinator.go index 1fc7873..9777440 100644 --- a/internal/growthreport/coordinator.go +++ b/internal/growthreport/coordinator.go @@ -34,8 +34,8 @@ func NewCoordinator(client *redis.Client, config CoordinatorConfig) *Coordinator func (c *Coordinator) AllowCreate(ctx context.Context, userID int64, now time.Time) error { return c.increment(ctx, fmt.Sprintf("growth:create:%d:%s", userID, now.UTC().Format("200601021504")), c.config.CreatePerMinute, time.Minute, now) } -func (c *Coordinator) AllowUncached(ctx context.Context, merchantID int64, now time.Time) error { - return c.increment(ctx, fmt.Sprintf("growth:scope:%d:%s", merchantID, tehranDate(now)), c.config.UncachedPerDay, 36*time.Hour, now) +func (c *Coordinator) AllowUncached(ctx context.Context, merchantID int64, terminalID string, now time.Time) error { + return c.increment(ctx, fmt.Sprintf("growth:scope:%d:%s:%s", merchantID, terminalID, tehranDate(now)), c.config.UncachedPerDay, 36*time.Hour, now) } func (c *Coordinator) ReserveBudget(ctx context.Context, requests, tokens int64, now time.Time) error { if c.config.DailyRequests < 1 || c.config.DailyTokens < 1 { diff --git a/internal/growthreport/coordinator_test.go b/internal/growthreport/coordinator_test.go index 1e176b4..12bfd3b 100644 --- a/internal/growthreport/coordinator_test.go +++ b/internal/growthreport/coordinator_test.go @@ -20,6 +20,15 @@ func TestCoordinatorFallsBackToBoundedProcessCounters(t *testing.T) { if err := coordinator.AllowCreate(context.Background(), 7, now); !errors.Is(err, ErrRateLimited) { t.Fatalf("third request error=%v", err) } + if err := coordinator.AllowUncached(context.Background(), 9, "terminal-a", now); err != nil { + t.Fatal(err) + } + if err := coordinator.AllowUncached(context.Background(), 9, "terminal-a", now); !errors.Is(err, ErrRateLimited) { + t.Fatalf("same-terminal uncached error=%v", err) + } + if err := coordinator.AllowUncached(context.Background(), 9, "terminal-b", now); err != nil { + t.Fatalf("another terminal should have an independent uncached quota: %v", err) + } if err := coordinator.ReserveBudget(context.Background(), MaxProviderRequests, 1600, now); !errors.Is(err, ErrBudgetExhausted) { t.Fatalf("budget without Redis error=%v", err) } diff --git a/internal/growthreport/orchestrator.go b/internal/growthreport/orchestrator.go index 76cb02d..866500b 100644 --- a/internal/growthreport/orchestrator.go +++ b/internal/growthreport/orchestrator.go @@ -37,9 +37,15 @@ func (s *Service) Create(ctx context.Context, userID, merchantID int64, request return Record{}, false, ErrDisabled } now := s.now().UTC() - if err := s.coordinator.AllowCreate(ctx, userID, now); err != nil { + hasAISummary, err := s.repository.HasAISummary(ctx, merchantID, request.TerminalID, growth.AnalyticsVersion) + if err != nil { return Record{}, false, err } + if hasAISummary { + if err := s.coordinator.AllowCreate(ctx, userID, now); err != nil { + return Record{}, false, err + } + } watermark, start, err := s.repository.DatasetMetadata(ctx) if err != nil { return Record{}, false, err @@ -51,11 +57,14 @@ func (s *Service) Create(ctx context.Context, userID, merchantID int64, request refreshDay = &day } record, cached, err := s.repository.CreateOrReuse(ctx, CreateRecord{ID: newReportID(), MerchantID: merchantID, TerminalID: request.TerminalID, Fingerprint: fingerprint, DatasetWatermark: watermark, ForecastStartDate: &start, CatalogVersion: growth.CatalogVersion, AnalyticsVersion: growth.AnalyticsVersion, PromptVersion: s.config.PromptVersion, SchemaVersion: s.config.SchemaVersion, Model: s.config.Model, ManualRefresh: request.Refresh, RefreshDay: refreshDay, IdempotencyKey: idempotencyKey, RequestHash: RequestHash(merchantID, request.TerminalID, request.Refresh)}, func() error { - return s.coordinator.AllowUncached(ctx, merchantID, now) + if !hasAISummary { + return nil + } + return s.coordinator.AllowUncached(ctx, merchantID, request.TerminalID, now) }) if err != nil { if errors.Is(err, ErrRefreshLimited) { - next, quotaErr := s.repository.NextRefreshAt(ctx, merchantID, request.TerminalID, now) + next, quotaErr := s.repository.NextRefreshAt(ctx, merchantID, request.TerminalID, growth.AnalyticsVersion, now) if quotaErr != nil { return Record{}, false, quotaErr } @@ -64,7 +73,7 @@ func (s *Service) Create(ctx context.Context, userID, merchantID int64, request return Record{}, false, err } record.Cached = cached - record.NextRefreshAt, err = s.repository.NextRefreshAt(ctx, merchantID, request.TerminalID, now) + record.NextRefreshAt, err = s.repository.NextRefreshAt(ctx, merchantID, request.TerminalID, growth.AnalyticsVersion, now) return record, cached, err } func (s *Service) Get(ctx context.Context, merchantID int64, id string) (Record, error) { @@ -74,7 +83,7 @@ func (s *Service) Get(ctx context.Context, merchantID int64, id string) (Record, if record.Scope.TerminalID != nil { terminalID = *record.Scope.TerminalID } - record.NextRefreshAt, _ = s.repository.NextRefreshAt(ctx, merchantID, terminalID, s.now()) + record.NextRefreshAt, _ = s.repository.NextRefreshAt(ctx, merchantID, terminalID, growth.AnalyticsVersion, s.now()) } return record, err } @@ -88,7 +97,7 @@ func (s *Service) Latest(ctx context.Context, merchantID int64, terminalID strin } else { record.Stale = record.Fingerprint != s.fingerprint(merchantID, terminalID, watermark, start) } - record.NextRefreshAt, _ = s.repository.NextRefreshAt(ctx, merchantID, terminalID, s.now()) + record.NextRefreshAt, _ = s.repository.NextRefreshAt(ctx, merchantID, terminalID, growth.AnalyticsVersion, s.now()) } return record, err } diff --git a/internal/growthreport/repository.go b/internal/growthreport/repository.go index 79bb84a..26e7349 100644 --- a/internal/growthreport/repository.go +++ b/internal/growthreport/repository.go @@ -95,7 +95,7 @@ func (r *Repository) CreateOrReuse(ctx context.Context, in CreateRecord, beforeI return Record{}, false, fmt.Errorf("locking growth report refresh: %w", err) } var hasAISummary bool - if err := tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM growth_reports WHERE merchant_id=$1 AND terminal_source_key IS NOT DISTINCT FROM $2::text AND status='READY' AND ai_response IS NOT NULL)`, in.MerchantID, nilIfEmpty(in.TerminalID)).Scan(&hasAISummary); err != nil { + if err := tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM growth_reports WHERE merchant_id=$1 AND terminal_source_key IS NOT DISTINCT FROM $2::text AND analytics_version=$3 AND status='READY' AND ai_response IS NOT NULL)`, in.MerchantID, nilIfEmpty(in.TerminalID), in.AnalyticsVersion).Scan(&hasAISummary); err != nil { return Record{}, false, fmt.Errorf("checking prior AI summary: %w", err) } var refreshUsed bool @@ -145,6 +145,14 @@ func (r *Repository) Latest(ctx context.Context, merchantID int64, terminalID, a return scanRecord(r.pool.QueryRow(ctx, reportSelect+` WHERE merchant_id=$1 AND terminal_source_key IS NOT DISTINCT FROM $2::text AND analytics_version=$3 AND status IN ('READY','PARTIAL') ORDER BY created_at DESC LIMIT 1`, merchantID, nilIfEmpty(terminalID), analyticsVersion)) } +func (r *Repository) HasAISummary(ctx context.Context, merchantID int64, terminalID, analyticsVersion string) (bool, error) { + var exists bool + if err := r.pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM growth_reports WHERE merchant_id=$1 AND terminal_source_key IS NOT DISTINCT FROM $2::text AND analytics_version=$3 AND status='READY' AND ai_response IS NOT NULL)`, merchantID, nilIfEmpty(terminalID), analyticsVersion).Scan(&exists); err != nil { + return false, fmt.Errorf("checking current AI summary: %w", err) + } + return exists, nil +} + func (r *Repository) Claim(ctx context.Context) (Record, error) { // PostgreSQL is the durable queue and coordination authority. SKIP LOCKED plus // the state transition prevents two worker processes from generating one row. @@ -263,13 +271,13 @@ func nilIfEmpty(value string) any { return value } -func (r *Repository) NextRefreshAt(ctx context.Context, merchantID int64, terminalID string, now time.Time) (time.Time, error) { +func (r *Repository) NextRefreshAt(ctx context.Context, merchantID int64, terminalID, analyticsVersion string, now time.Time) (time.Time, error) { var hasAISummary bool var last *time.Time if err := r.pool.QueryRow(ctx, `SELECT - EXISTS(SELECT 1 FROM growth_reports WHERE merchant_id=$1 AND terminal_source_key IS NOT DISTINCT FROM $2::text AND status='READY' AND ai_response IS NOT NULL), + EXISTS(SELECT 1 FROM growth_reports WHERE merchant_id=$1 AND terminal_source_key IS NOT DISTINCT FROM $2::text AND analytics_version=$3 AND status='READY' AND ai_response IS NOT NULL), max(created_at) FILTER (WHERE manual_refresh=true) - FROM growth_reports WHERE merchant_id=$1 AND terminal_source_key IS NOT DISTINCT FROM $2::text`, merchantID, nilIfEmpty(terminalID)).Scan(&hasAISummary, &last); err != nil { + FROM growth_reports WHERE merchant_id=$1 AND terminal_source_key IS NOT DISTINCT FROM $2::text AND analytics_version=$3`, merchantID, nilIfEmpty(terminalID), analyticsVersion).Scan(&hasAISummary, &last); err != nil { return time.Time{}, fmt.Errorf("loading refresh quota: %w", err) } if !hasAISummary || last == nil { diff --git a/internal/growthreport/repository_integration_test.go b/internal/growthreport/repository_integration_test.go index 591356b..d9891bb 100644 --- a/internal/growthreport/repository_integration_test.go +++ b/internal/growthreport/repository_integration_test.go @@ -70,7 +70,7 @@ func TestRepositoryAllowsManualRefreshUntilAISummaryExists(t *testing.T) { if err != nil { t.Fatal(err) } - if next, err := repository.NextRefreshAt(ctx, merchantID, "", start); err != nil || next.After(start) { + if next, err := repository.NextRefreshAt(ctx, merchantID, "", "test", start); err != nil || next.After(start) { t.Fatalf("refresh without AI summary should remain available: next=%s error=%v", next, err) } second, _, err := repository.CreateOrReuse(ctx, newInput("growth_bbbbbbbbbbbbbbbbbbbbbbbb", "second"), nil) @@ -84,7 +84,7 @@ func TestRepositoryAllowsManualRefreshUntilAISummaryExists(t *testing.T) { if !errors.Is(err, ErrRefreshLimited) { t.Fatalf("refresh after AI summary error=%v, want ErrRefreshLimited (first=%s)", err, first.ReportID) } - if next, err := repository.NextRefreshAt(ctx, merchantID, "", start); err != nil || !next.After(start) { + if next, err := repository.NextRefreshAt(ctx, merchantID, "", "test", start); err != nil || !next.After(start) { t.Fatalf("refresh after AI summary should have a future boundary: next=%s error=%v", next, err) } } @@ -103,6 +103,9 @@ func TestRepositoryLatestIgnoresReportsFromOlderAnalyticsVersions(t *testing.T) if _, err := repository.Latest(ctx, merchantID, "", "1.2.0"); !errors.Is(err, ErrNotFound) { t.Fatalf("Latest error=%v, want ErrNotFound for a stale analytics version", err) } + if hasSummary, err := repository.HasAISummary(ctx, merchantID, "", "1.2.0"); err != nil || hasSummary { + t.Fatalf("stale analytics versions must not count as a visible AI summary: hasSummary=%t error=%v", hasSummary, err) + } } func TestRepositoryRefreshQuotaIsScopedPerTerminal(t *testing.T) {