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

feat: harden growth report orchestration

Browse Source
This commit is contained in:
Amirhossein Khalili
2026-08-21 00:43:33 +03:30
parent dd0ae15a95
commit d00700ffb8
5 changed files with 54 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
internal/growthreport/adapter.go
View File

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

8
internal/growthreport/avalai.go
View File

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

20
internal/growthreport/coordinator.go
View File

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

7
internal/growthreport/coordinator_test.go
View File

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

21
internal/growthreport/orchestrator.go
View File

@@ -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, &copy, Usage{}, failure)
}