diff --git a/internal/growthreport/adapter.go b/internal/growthreport/adapter.go index 679af1f..0ee5fcb 100644 --- a/internal/growthreport/adapter.go +++ b/internal/growthreport/adapter.go @@ -59,8 +59,11 @@ func (b *PostgreSQLBuilder) Build(ctx context.Context, merchantID int64, termina if err := b.pool.QueryRow(ctx, `SELECT category_title FROM merchants WHERE id=$1`, merchantID).Scan(&category); err != nil { return Analysis{}, fmt.Errorf("loading merchant category: %w", err) } - rows, err := b.pool.Query(ctx, `SELECT (ps.created_at AT TIME ZONE 'Asia/Tehran')::date,count(*),count(*) FILTER(WHERE ps.status IN ('Paid','Verified')),coalesce(sum(ps.amount_rial) FILTER(WHERE ps.status IN ('Paid','Verified')),0) - FROM payment_sessions ps WHERE ps.merchant_id=$1 AND ($2='' OR ps.terminal_id=(SELECT id FROM terminals WHERE merchant_id=$1 AND source_key=$2)) GROUP BY 1 ORDER BY 1`, merchantID, terminalID) + rows, err := b.pool.Query(ctx, `WITH bounds AS (SELECT min(created_at AT TIME ZONE 'Asia/Tehran')::date first_day,max(created_at AT TIME ZONE 'Asia/Tehran')::date last_day FROM payment_sessions), + days AS (SELECT generate_series(first_day,last_day,interval '1 day')::date local_date FROM bounds), scoped AS ( + SELECT (ps.created_at AT TIME ZONE 'Asia/Tehran')::date local_date,count(*) sessions,count(*) FILTER(WHERE ps.status IN ('Paid','Verified')) paid_sessions,coalesce(sum(ps.amount_rial) FILTER(WHERE ps.status IN ('Paid','Verified')),0) paid_amount + FROM payment_sessions ps WHERE ps.merchant_id=$1 AND ($2='' OR ps.terminal_id=(SELECT id FROM terminals WHERE merchant_id=$1 AND source_key=$2)) GROUP BY 1) + SELECT d.local_date,coalesce(s.sessions,0),coalesce(s.paid_sessions,0),coalesce(s.paid_amount,0) FROM days d LEFT JOIN scoped s USING(local_date) ORDER BY d.local_date`, merchantID, terminalID) if err != nil { return Analysis{}, fmt.Errorf("loading growth history: %w", err) } @@ -151,7 +154,7 @@ func (b *PostgreSQLBuilder) loadPeer(ctx context.Context, merchantID int64, term func convertFallback(items []growth.CampaignBrief) AICopy { result := AICopy{Briefs: make([]CampaignBrief, 0, len(items))} for _, item := range items { - result.Briefs = append(result.Briefs, CampaignBrief{Headline: item.HeadlineFA, Interpretation: item.GuidanceFA, OfferSuggestion: item.OfferFA, PreparationAction: item.PreparationFA, Limitation: item.LimitationFA, EvidenceIDs: item.EvidenceIDs}) + result.Briefs = append(result.Briefs, CampaignBrief{Headline: item.HeadlineFA, Interpretation: item.GuidanceFA, OfferSuggestion: item.OfferFA, MessageSuggestion: item.MessageFA, PreparationAction: item.PreparationFA, Limitation: item.LimitationFA, EvidenceIDs: item.EvidenceIDs}) } return result } @@ -166,6 +169,9 @@ func (t EvidenceTools) Execute(_ context.Context, name string, arguments json.Ra if err := json.Unmarshal(arguments, &args); err != nil { return nil, fmt.Errorf("invalid tool arguments: %w", err) } + if len(args.EvidenceID) > 128 { + return nil, fmt.Errorf("growth tool: invalid evidence id") + } switch name { case "get_growth_context": return json.Marshal(map[string]any{"confidence": t.analysis.Evidence.Forecast.Confidence, "opportunityEvidenceIds": evidenceIDs(t.analysis.Evidence.Opportunities), "riskEvidenceIds": riskEvidenceIDs(t.analysis.Evidence.Risks)}) diff --git a/internal/growthreport/avalai.go b/internal/growthreport/avalai.go index a8864e1..e1e629a 100644 --- a/internal/growthreport/avalai.go +++ b/internal/growthreport/avalai.go @@ -276,7 +276,13 @@ func growthTools() []map[string]any { names := []string{"get_growth_context", "get_event_evidence", "get_peer_context", "get_campaign_playbook"} tools := make([]map[string]any, 0, len(names)) for _, name := range names { - tools = append(tools, map[string]any{"type": "function", "name": name, "description": "Return approved sanitized deterministic evidence for this authenticated report scope.", "strict": true, "parameters": map[string]any{"type": "object", "properties": map[string]any{"evidenceId": map[string]any{"type": "string"}}, "required": []string{"evidenceId"}, "additionalProperties": false}}) + properties := map[string]any{} + required := []string{} + if name == "get_event_evidence" || name == "get_campaign_playbook" { + properties["evidenceId"] = map[string]any{"type": "string", "minLength": 1, "maxLength": 128} + required = []string{"evidenceId"} + } + tools = append(tools, map[string]any{"type": "function", "name": name, "description": "Return approved sanitized deterministic evidence for this authenticated report scope.", "strict": true, "parameters": map[string]any{"type": "object", "properties": properties, "required": required, "additionalProperties": false}}) } return tools } diff --git a/internal/growthreport/coordinator.go b/internal/growthreport/coordinator.go index abaaf35..18cdcd8 100644 --- a/internal/growthreport/coordinator.go +++ b/internal/growthreport/coordinator.go @@ -42,12 +42,27 @@ func (c *Coordinator) ReserveBudget(ctx context.Context, tokens int64, now time. if c.config.DailyRequests < 1 || c.config.DailyTokens < 1 { return ErrBudgetExhausted } - if err := c.increment(ctx, "growth:budget:req:"+tehranDate(now), c.config.DailyRequests, 36*time.Hour, now); err != nil { + if c.redis == nil { return ErrBudgetExhausted } - if err := c.incrementBy(ctx, "growth:budget:tok:"+tehranDate(now), tokens, c.config.DailyTokens, 36*time.Hour, now); err != nil { + if err := c.incrementRedis(ctx, "growth:budget:req:"+tehranDate(now), 1, c.config.DailyRequests, 36*time.Hour); err != nil { return ErrBudgetExhausted } + if err := c.incrementRedis(ctx, "growth:budget:tok:"+tehranDate(now), tokens, c.config.DailyTokens, 36*time.Hour); err != nil { + return ErrBudgetExhausted + } + return nil +} + +func (c *Coordinator) incrementRedis(ctx context.Context, key string, amount, limit int64, ttl time.Duration) error { + script := redis.NewScript(`local v=redis.call('INCRBY',KEYS[1],ARGV[1]);if v==tonumber(ARGV[1]) then redis.call('PEXPIRE',KEYS[1],ARGV[2]) end;if v>tonumber(ARGV[3]) then return 0 end;return 1`) + value, err := script.Run(ctx, c.redis, []string{key}, amount, ttl.Milliseconds(), limit).Int() + if err != nil { + return err + } + if value == 0 { + return ErrRateLimited + } return nil } func (c *Coordinator) Enter(ctx context.Context) (func(), error) { @@ -63,6 +78,7 @@ func (c *Coordinator) Publish(ctx context.Context, reportID string) { _ = c.redis.LPush(ctx, "growth:jobs", reportID).Err() } } + func (c *Coordinator) increment(ctx context.Context, key string, limit int64, ttl time.Duration, now time.Time) error { return c.incrementBy(ctx, key, 1, limit, ttl, now) } diff --git a/internal/growthreport/coordinator_test.go b/internal/growthreport/coordinator_test.go index 95ac06e..9bd4e16 100644 --- a/internal/growthreport/coordinator_test.go +++ b/internal/growthreport/coordinator_test.go @@ -20,10 +20,7 @@ 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.ReserveBudget(context.Background(), 1600, now); err != nil { - t.Fatal(err) - } - if err := coordinator.ReserveBudget(context.Background(), 1, now); !errors.Is(err, ErrBudgetExhausted) { - t.Fatalf("second budget error=%v", err) + if err := coordinator.ReserveBudget(context.Background(), 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 406c276..993f938 100644 --- a/internal/growthreport/orchestrator.go +++ b/internal/growthreport/orchestrator.go @@ -7,6 +7,7 @@ import ( "encoding/json" "errors" "fmt" + "log/slog" "time" "git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/growth" @@ -41,7 +42,7 @@ func (s *Service) Create(ctx context.Context, userID, merchantID int64, request if err != nil { return Record{}, false, err } - fingerprint := growth.Fingerprint(growth.FingerprintInput{MerchantID: merchantID, TerminalID: request.TerminalID, DatasetWatermark: watermark, ForecastStart: start, HorizonDays: 90, CatalogVersion: growth.CatalogVersion, AnalyticsVersion: growth.AnalyticsVersion, PromptVersion: s.config.PromptVersion, SchemaVersion: s.config.SchemaVersion, Model: s.config.Model}) + fingerprint := growth.Fingerprint(growth.FingerprintInput{MerchantID: merchantID, TerminalID: request.TerminalID, DatasetWatermark: watermark, ForecastStart: start, HorizonDays: 90, CatalogVersion: growth.CatalogVersion, AnalyticsVersion: growth.AnalyticsVersion, PromptVersion: s.config.PromptVersion, SchemaVersion: s.config.SchemaVersion, Model: s.config.Model, ModelConfigVersion: growth.DefaultModelConfigVersion}) var refreshDay *time.Time if request.Refresh { day := tehranMidnight(now) @@ -49,6 +50,13 @@ func (s *Service) Create(ctx context.Context, userID, merchantID int64, request } 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)}) if err != nil { + if errors.Is(err, ErrRefreshLimited) { + next, quotaErr := s.repository.NextRefreshAt(ctx, merchantID, now) + if quotaErr != nil { + return Record{}, false, quotaErr + } + return Record{}, false, &RefreshLimitError{NextRefreshAt: next} + } return Record{}, false, err } if !cached { @@ -109,22 +117,29 @@ func (p *Processor) ProcessNext(ctx context.Context) error { } copy := analysis.Fallback status := StatusPartial - failure := "AI_SKIPPED" + failure := "AI_DISABLED_OR_UNAVAILABLE" if analysis.ShouldCallAI && p.aiEnabled && p.avalai != nil { - if err := p.coordinator.ReserveBudget(ctx, 1600, p.now()); err == nil { + reservedTokens := int64((len(analysis.Encoded)+3)/4*2 + 1600) + if err := p.coordinator.ReserveBudget(ctx, reservedTokens, p.now()); err == nil { release, enterErr := p.coordinator.Enter(ctx) if enterErr == nil { generated, usage, aiErr := p.avalai.GenerateWithTools(ctx, analysis.Encoded, analysis.KnownEvidence, NewEvidenceTools(analysis)) release() if aiErr == nil { + slog.Info("growth AI generation complete", "report_id", record.ReportID, "input_tokens", usage.InputTokens, "output_tokens", usage.OutputTokens, "latency_ms", usage.LatencyMS) return p.repository.Complete(ctx, record.ReportID, StatusReady, analysis.Encoded, &generated, usage, "") } + slog.Warn("growth AI output rejected", "report_id", record.ReportID, "reason", "AI_REJECTED") failure = "AI_REJECTED" } } else { failure = "AI_BUDGET_EXHAUSTED" + slog.Warn("growth AI skipped", "report_id", record.ReportID, "reason", failure) } } + if !analysis.ShouldCallAI || !p.aiEnabled { + slog.Info("growth deterministic fallback used", "report_id", record.ReportID, "reason", failure) + } return p.repository.Complete(ctx, record.ReportID, status, analysis.Encoded, ©, Usage{}, failure) }