starcoach-hackathon/zarinpal-analytics-backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: unblock first growth report generation

Browse Source
This commit is contained in:
Amirhossein Khalili
2026-08-21 07:49:04 +08:00
parent e6afaa7a10
commit 927319807b
7 changed files with 67 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
internal/growthreport/coordinator.go
View File

@@ -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 {

9
internal/growthreport/coordinator_test.go
View File

@@ -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)
}

21
internal/growthreport/orchestrator.go
View File

@@ -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
}

16
internal/growthreport/repository.go
View File

@@ -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 {

7
internal/growthreport/repository_integration_test.go
View File

@@ -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) {