diff --git a/internal/growthreport/adapter.go b/internal/growthreport/adapter.go new file mode 100644 index 0000000..679af1f --- /dev/null +++ b/internal/growthreport/adapter.go @@ -0,0 +1,213 @@ +package growthreport + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/growth" + "github.com/jackc/pgx/v5/pgxpool" +) + +type HistoricalWin struct { + growth.EventEffect + EventTitleFA string `json:"eventTitleFa"` + StartDate time.Time `json:"startDate"` + EndDate time.Time `json:"endDate"` +} + +type PeerEvidence struct { + State growth.PeerState `json:"state"` + Reason growth.SuppressionReason `json:"reason,omitempty"` + EligiblePeerCount int `json:"eligiblePeerCount"` + CombinedPeerSessionCount int64 `json:"combinedPeerSessionCount"` + MedianPaidSessions int64 `json:"medianPaidSessions"` + PercentileBPS int64 `json:"percentileBps"` +} + +type Evidence struct { + Forecast growth.Forecast `json:"forecast"` + HistoricalWins []HistoricalWin `json:"historicalWins"` + Opportunities []growth.Opportunity `json:"opportunities"` + Risks []growth.Risk `json:"risks"` + Peer *PeerEvidence `json:"peer,omitempty"` + FallbackBriefs []growth.CampaignBrief `json:"fallbackBriefs"` +} + +type Analysis struct { + Evidence Evidence + Encoded json.RawMessage + KnownEvidence map[string]struct{} + Fallback AICopy + ShouldCallAI bool +} + +type Builder interface { + Build(ctx context.Context, merchantID int64, terminalID string) (Analysis, error) +} + +type PostgreSQLBuilder struct{ pool *pgxpool.Pool } + +func NewPostgreSQLBuilder(pool *pgxpool.Pool) *PostgreSQLBuilder { + return &PostgreSQLBuilder{pool: pool} +} + +func (b *PostgreSQLBuilder) Build(ctx context.Context, merchantID int64, terminalID string) (Analysis, error) { + catalog := growth.Catalog1405() + var category string + 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) + if err != nil { + return Analysis{}, fmt.Errorf("loading growth history: %w", err) + } + defer rows.Close() + history := make([]growth.DailyPoint, 0, 366) + for rows.Next() { + var point growth.DailyPoint + if err := rows.Scan(&point.Date, &point.SessionCount, &point.PaidSessionCount, &point.PaidAmountRial); err != nil { + return Analysis{}, fmt.Errorf("scanning growth history: %w", err) + } + for _, event := range catalog { + if !point.Date.Before(event.StartDate) && !point.Date.After(event.EndDate) { + point.EventIDs = append(point.EventIDs, event.ID) + } + } + history = append(history, point) + } + if err := rows.Err(); err != nil { + return Analysis{}, fmt.Errorf("iterating growth history: %w", err) + } + var latest time.Time + if err := b.pool.QueryRow(ctx, `SELECT max(created_at AT TIME ZONE 'Asia/Tehran')::date FROM payment_sessions`).Scan(&latest); err != nil { + return Analysis{}, fmt.Errorf("loading dataset watermark date: %w", err) + } + forecastStart := latest.AddDate(0, 0, 1) + peer, err := b.loadPeer(ctx, merchantID, terminalID) + if err != nil { + return Analysis{}, err + } + deterministic := growth.BuildDeterministicReport(growth.ReportInput{History: history, Catalog: catalog, ForecastStart: forecastStart, HorizonDays: 90, Category: category, Peer: &peer}) + events := make(map[string]growth.Event, len(catalog)) + for _, event := range catalog { + events[event.ID] = event + } + evidence := Evidence{Forecast: deterministic.Forecast, Opportunities: deterministic.Opportunities, Risks: deterministic.Risks, FallbackBriefs: deterministic.FallbackBriefs} + for _, effect := range deterministic.HistoricalWins { + event := events[effect.EventID] + evidence.HistoricalWins = append(evidence.HistoricalWins, HistoricalWin{EventEffect: effect, EventTitleFA: event.TitleFA, StartDate: event.StartDate, EndDate: event.EndDate}) + } + if deterministic.Peer != nil { + evidence.Peer = &PeerEvidence{State: deterministic.Peer.State, Reason: deterministic.Peer.Reason, EligiblePeerCount: deterministic.Peer.EligiblePeerCount, CombinedPeerSessionCount: deterministic.Peer.CombinedPeerSessions, MedianPaidSessions: deterministic.Peer.MedianPaidSessions, PercentileBPS: deterministic.Peer.PercentileBPS} + } + encoded, err := json.Marshal(evidence) + if err != nil { + return Analysis{}, fmt.Errorf("encoding deterministic growth evidence: %w", err) + } + known := map[string]struct{}{} + for _, win := range evidence.HistoricalWins { + known[win.EvidenceID] = struct{}{} + } + for _, opportunity := range evidence.Opportunities { + known[opportunity.EvidenceID] = struct{}{} + } + for _, risk := range evidence.Risks { + known[risk.EvidenceID] = struct{}{} + } + return Analysis{Evidence: evidence, Encoded: encoded, KnownEvidence: known, Fallback: convertFallback(deterministic.FallbackBriefs), ShouldCallAI: deterministic.Forecast.Confidence != growth.ConfidenceInsufficientData && (len(deterministic.Opportunities) > 0 || len(deterministic.Risks) > 0)}, nil +} + +func (b *PostgreSQLBuilder) loadPeer(ctx context.Context, merchantID int64, terminalID string) (growth.PeerInput, error) { + if terminalID != "" { + return growth.PeerInput{TerminalScoped: true}, nil + } + var own int64 + if err := b.pool.QueryRow(ctx, `SELECT count(*) FROM payment_sessions WHERE merchant_id=$1 AND status IN ('Paid','Verified')`, merchantID).Scan(&own); err != nil { + return growth.PeerInput{}, fmt.Errorf("loading merchant peer value: %w", err) + } + rows, err := b.pool.Query(ctx, `SELECT ps.merchant_id,count(*) FILTER(WHERE ps.status IN ('Paid','Verified')),count(*) FROM payment_sessions ps JOIN merchants m ON m.id=ps.merchant_id WHERE m.category_id=(SELECT category_id FROM merchants WHERE id=$1) AND ps.merchant_id<>$1 GROUP BY ps.merchant_id`, merchantID) + if err != nil { + return growth.PeerInput{}, fmt.Errorf("loading peer values: %w", err) + } + defer rows.Close() + input := growth.PeerInput{MerchantPaidSessions: own} + for rows.Next() { + var peer growth.PeerMerchantValue + var id int64 + if err := rows.Scan(&id, &peer.PaidSessions, &peer.SessionCount); err != nil { + return growth.PeerInput{}, fmt.Errorf("scanning peer value: %w", err) + } + input.Peers = append(input.Peers, peer) + } + if err := rows.Err(); err != nil { + return growth.PeerInput{}, fmt.Errorf("iterating peer values: %w", err) + } + return input, nil +} + +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}) + } + return result +} + +type EvidenceTools struct{ analysis Analysis } + +func NewEvidenceTools(analysis Analysis) EvidenceTools { return EvidenceTools{analysis: analysis} } +func (t EvidenceTools) Execute(_ context.Context, name string, arguments json.RawMessage) (json.RawMessage, error) { + var args struct { + EvidenceID string `json:"evidenceId"` + } + if err := json.Unmarshal(arguments, &args); err != nil { + return nil, fmt.Errorf("invalid tool arguments: %w", err) + } + 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)}) + case "get_peer_context": + if t.analysis.Evidence.Peer == nil { + return []byte(`{"state":"SUPPRESSED"}`), nil + } + return json.Marshal(t.analysis.Evidence.Peer) + case "get_event_evidence": + for _, item := range t.analysis.Evidence.HistoricalWins { + if item.EvidenceID == args.EvidenceID { + return json.Marshal(item) + } + } + for _, item := range t.analysis.Evidence.Opportunities { + if item.EvidenceID == args.EvidenceID { + return json.Marshal(item) + } + } + case "get_campaign_playbook": + for _, item := range t.analysis.Fallback.Briefs { + for _, id := range item.EvidenceIDs { + if id == args.EvidenceID { + return json.Marshal(item) + } + } + } + } + return nil, errorsNewUnknownEvidence() +} +func evidenceIDs(items []growth.Opportunity) []string { + result := make([]string, 0, len(items)) + for _, item := range items { + result = append(result, item.EvidenceID) + } + return result +} +func riskEvidenceIDs(items []growth.Risk) []string { + result := make([]string, 0, len(items)) + for _, item := range items { + result = append(result, item.EvidenceID) + } + return result +} +func errorsNewUnknownEvidence() error { return fmt.Errorf("growth tool: unknown evidence") } diff --git a/internal/growthreport/avalai.go b/internal/growthreport/avalai.go index b35ebf4..a8864e1 100644 --- a/internal/growthreport/avalai.go +++ b/internal/growthreport/avalai.go @@ -17,94 +17,275 @@ const maxAvalAIResponseBytes = 1 << 20 type AvalAIConfig struct { BaseURL string - APIKey string - Model string + APIKey string + Model string Timeout time.Duration } -type AvalAIClient struct { config AvalAIConfig; client *http.Client } +type AvalAIClient struct { + config AvalAIConfig + client *http.Client +} + +type GrowthAnalyticsTools interface { + Execute(ctx context.Context, name string, arguments json.RawMessage) (json.RawMessage, error) +} func NewAvalAIClient(config AvalAIConfig, client *http.Client) *AvalAIClient { - if config.BaseURL == "" { config.BaseURL = "https://api.avalai.ir" } - if config.Timeout <= 0 || config.Timeout > 30*time.Second { config.Timeout = 30*time.Second } - if client == nil { client = &http.Client{} } + if config.BaseURL == "" { + config.BaseURL = "https://api.avalai.ir" + } + if config.Timeout <= 0 || config.Timeout > 30*time.Second { + config.Timeout = 30 * time.Second + } + if client == nil { + client = &http.Client{} + } return &AvalAIClient{config: config, client: client} } func (c *AvalAIClient) Generate(ctx context.Context, evidence json.RawMessage, known map[string]struct{}) (AICopy, Usage, error) { - if c.config.APIKey == "" || c.config.Model == "" { return AICopy{}, Usage{}, errors.New("avalai: disabled") } - body := map[string]any{ + return c.GenerateWithTools(ctx, evidence, known, nil) +} + +func (c *AvalAIClient) GenerateWithTools(ctx context.Context, evidence json.RawMessage, known map[string]struct{}, executor GrowthAnalyticsTools) (AICopy, Usage, error) { + if c.config.APIKey == "" || c.config.Model == "" { + return AICopy{}, Usage{}, errors.New("avalai: disabled") + } + base := map[string]any{ "model": c.config.Model, "store": false, "parallel_tool_calls": false, "max_output_tokens": 1600, "max_tool_calls": 4, "instructions": "شما تحلیل‌گر رشد فارسی هستید. فقط بر اساس شواهد ارائه‌شده پیشنهاد کیفی بدهید. هیچ عدد، تاریخ یا ادعای تازه‌ای ننویسید. شناسه شواهد را دقیق ارجاع دهید.", - "input": string(evidence), "tools": growthTools(), - "text": map[string]any{"format": map[string]any{"type":"json_schema", "name":"growth_copy", "strict":true, "schema": aiCopySchema()}}, + "tools": growthTools(), + "text": map[string]any{"format": map[string]any{"type": "json_schema", "name": "growth_copy", "strict": true, "schema": aiCopySchema()}}, } - encoded, err := json.Marshal(body); if err != nil { return AICopy{}, Usage{}, fmt.Errorf("encoding AvalAI request: %w", err) } var last error for attempt := 0; attempt < 2; attempt++ { - copy, usage, transient, err := c.do(ctx, encoded, known) - if err == nil { return copy, usage, nil } + copy, usage, transient, err := c.workflow(ctx, base, evidence, known, executor) + if err == nil { + return copy, usage, nil + } last = err - if !transient || ctx.Err() != nil { break } + if !transient || ctx.Err() != nil { + break + } } return AICopy{}, Usage{}, last } -func (c *AvalAIClient) do(parent context.Context, body []byte, known map[string]struct{}) (AICopy, Usage, bool, error) { - ctx, cancel := context.WithTimeout(parent, c.config.Timeout); defer cancel() +type responseEnvelope struct { + Status string `json:"status"` + Output []json.RawMessage `json:"output"` + Usage struct { + InputTokens int64 `json:"input_tokens"` + OutputTokens int64 `json:"output_tokens"` + } `json:"usage"` +} + +type outputItem struct { + Type string `json:"type"` + CallID string `json:"call_id"` + Name string `json:"name"` + Arguments json.RawMessage `json:"arguments"` + Content []struct { + Type string `json:"type"` + Text string `json:"text"` + Refusal string `json:"refusal"` + } `json:"content"` +} + +func (c *AvalAIClient) workflow(ctx context.Context, base map[string]any, evidence json.RawMessage, known map[string]struct{}, executor GrowthAnalyticsTools) (AICopy, Usage, bool, error) { + firstBody := cloneRequest(base) + firstBody["input"] = string(evidence) + first, latency, transient, err := c.do(ctx, firstBody) + if err != nil { + return AICopy{}, Usage{}, transient, err + } + calls := make([]outputItem, 0, 4) + for _, raw := range first.Output { + var item outputItem + if json.Unmarshal(raw, &item) == nil && item.Type == "function_call" { + calls = append(calls, item) + } + } + usage := Usage{InputTokens: first.Usage.InputTokens, OutputTokens: first.Usage.OutputTokens, LatencyMS: latency} + final := first + if len(calls) > 0 { + if executor == nil { + return AICopy{}, usage, false, errors.New("avalai: unexpected tool call") + } + if len(calls) > 4 { + return AICopy{}, usage, false, errors.New("avalai: tool call limit exceeded") + } + input := make([]any, 0, len(first.Output)+len(calls)+1) + input = append(input, map[string]any{"role": "user", "content": string(evidence)}) + for _, raw := range first.Output { + var value any + if json.Unmarshal(raw, &value) == nil { + input = append(input, value) + } + } + for _, call := range calls { + if !allowedTool(call.Name) { + return AICopy{}, usage, false, errors.New("avalai: unknown tool") + } + result, err := executor.Execute(ctx, call.Name, call.Arguments) + if err != nil { + return AICopy{}, usage, false, fmt.Errorf("executing growth tool: %w", err) + } + input = append(input, map[string]any{"type": "function_call_output", "call_id": call.CallID, "output": string(result)}) + } + secondBody := cloneRequest(base) + secondBody["input"] = input + second, secondLatency, secondTransient, err := c.do(ctx, secondBody) + if err != nil { + return AICopy{}, usage, secondTransient, err + } + usage.InputTokens += second.Usage.InputTokens + usage.OutputTokens += second.Usage.OutputTokens + usage.LatencyMS += secondLatency + final = second + for _, raw := range second.Output { + var item outputItem + if json.Unmarshal(raw, &item) == nil && item.Type == "function_call" { + return AICopy{}, usage, false, errors.New("avalai: model exceeded two turns") + } + } + } + copy, err := parseAICopy(final, known) + if err != nil { + return AICopy{}, usage, false, err + } + return copy, usage, false, nil +} + +func (c *AvalAIClient) do(parent context.Context, bodyValue map[string]any) (responseEnvelope, int64, bool, error) { + body, err := json.Marshal(bodyValue) + if err != nil { + return responseEnvelope{}, 0, false, fmt.Errorf("encoding AvalAI request: %w", err) + } + ctx, cancel := context.WithTimeout(parent, c.config.Timeout) + defer cancel() req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(c.config.BaseURL, "/")+"/v1/responses", bytes.NewReader(body)) - if err != nil { return AICopy{}, Usage{}, false, fmt.Errorf("creating AvalAI request: %w", err) } - req.Header.Set("Authorization", "Bearer "+c.config.APIKey); req.Header.Set("Content-Type", "application/json") - started := time.Now(); resp, err := c.client.Do(req) - if err != nil { return AICopy{}, Usage{}, true, fmt.Errorf("calling AvalAI: %w", err) } + if err != nil { + return responseEnvelope{}, 0, false, fmt.Errorf("creating AvalAI request: %w", err) + } + req.Header.Set("Authorization", "Bearer "+c.config.APIKey) + req.Header.Set("Content-Type", "application/json") + started := time.Now() + resp, err := c.client.Do(req) + if err != nil { + return responseEnvelope{}, 0, true, fmt.Errorf("calling AvalAI: %w", err) + } defer resp.Body.Close() responseBody, err := io.ReadAll(io.LimitReader(resp.Body, maxAvalAIResponseBytes+1)) - if err != nil { return AICopy{}, Usage{}, true, fmt.Errorf("reading AvalAI response: %w", err) } - if len(responseBody) > maxAvalAIResponseBytes { return AICopy{}, Usage{}, false, errors.New("avalai: response too large") } - if resp.StatusCode < 200 || resp.StatusCode >= 300 { return AICopy{}, Usage{}, resp.StatusCode == 429 || resp.StatusCode >= 500, fmt.Errorf("avalai: HTTP %d", resp.StatusCode) } - var result struct { - Status string `json:"status"` - Output []struct { Type string `json:"type"`; Content []struct { Type string `json:"type"`; Text string `json:"text"`; Refusal string `json:"refusal"` } `json:"content"` } `json:"output"` - Usage struct { InputTokens int64 `json:"input_tokens"`; OutputTokens int64 `json:"output_tokens"` } `json:"usage"` + if err != nil { + return responseEnvelope{}, 0, true, fmt.Errorf("reading AvalAI response: %w", err) } - if err := json.Unmarshal(responseBody, &result); err != nil { return AICopy{}, Usage{}, false, fmt.Errorf("decoding AvalAI response: %w", err) } - if result.Status != "completed" { return AICopy{}, Usage{}, false, fmt.Errorf("avalai: incomplete response status %q", result.Status) } + if len(responseBody) > maxAvalAIResponseBytes { + return responseEnvelope{}, 0, false, errors.New("avalai: response too large") + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return responseEnvelope{}, 0, resp.StatusCode == 429 || resp.StatusCode >= 500, fmt.Errorf("avalai: HTTP %d", resp.StatusCode) + } + var result responseEnvelope + if err := json.Unmarshal(responseBody, &result); err != nil { + return responseEnvelope{}, 0, false, fmt.Errorf("decoding AvalAI response: %w", err) + } + if result.Status != "completed" { + return responseEnvelope{}, 0, false, fmt.Errorf("avalai: incomplete response status %q", result.Status) + } + return result, time.Since(started).Milliseconds(), false, nil +} + +func parseAICopy(result responseEnvelope, known map[string]struct{}) (AICopy, error) { var text string - for _, item := range result.Output { if item.Type == "message" { for _, content := range item.Content { if content.Refusal != "" { return AICopy{}, Usage{}, false, errors.New("avalai: refused") }; if content.Type == "output_text" { text += content.Text } } } } + for _, raw := range result.Output { + var item outputItem + if json.Unmarshal(raw, &item) == nil && item.Type == "message" { + for _, content := range item.Content { + if content.Refusal != "" { + return AICopy{}, errors.New("avalai: refused") + } + if content.Type == "output_text" { + text += content.Text + } + } + } + } var copy AICopy - if text == "" || json.Unmarshal([]byte(text), ©) != nil { return AICopy{}, Usage{}, false, errors.New("avalai: invalid structured output") } - if err := ValidateAICopy(copy, known); err != nil { return AICopy{}, Usage{}, false, err } - return copy, Usage{InputTokens: result.Usage.InputTokens, OutputTokens: result.Usage.OutputTokens, LatencyMS: time.Since(started).Milliseconds()}, false, nil + if text == "" || json.Unmarshal([]byte(text), ©) != nil { + return AICopy{}, errors.New("avalai: invalid structured output") + } + if err := ValidateAICopy(copy, known); err != nil { + return AICopy{}, err + } + return copy, nil +} + +func cloneRequest(base map[string]any) map[string]any { + result := make(map[string]any, len(base)+1) + for key, value := range base { + result[key] = value + } + return result +} +func allowedTool(name string) bool { + switch name { + case "get_growth_context", "get_event_evidence", "get_peer_context", "get_campaign_playbook": + return true + } + return false } func ValidateAICopy(copy AICopy, known map[string]struct{}) error { - if len(copy.Briefs) == 0 || len(copy.Briefs) > 5 { return errors.New("avalai: invalid brief count") } + if len(copy.Briefs) == 0 || len(copy.Briefs) > 5 { + return errors.New("avalai: invalid brief count") + } for _, brief := range copy.Briefs { fields := []string{brief.Headline, brief.Interpretation, brief.OfferSuggestion, brief.MessageSuggestion, brief.PreparationAction, brief.Limitation} for _, field := range fields { - if len([]rune(field)) > 600 { return errors.New("avalai: field too long") } - for _, r := range field { if unicode.IsDigit(r) { return errors.New("avalai: numeric claim rejected") } } + if len([]rune(field)) > 600 { + return errors.New("avalai: field too long") + } + for _, r := range field { + if unicode.IsDigit(r) { + return errors.New("avalai: numeric claim rejected") + } + } lower := strings.ToLower(field) - for _, forbidden := range []string{"payer", "card", "رمز", "شماره کارت", "sql", "merchant_id"} { if strings.Contains(lower, forbidden) { return errors.New("avalai: unsafe content rejected") } } + for _, forbidden := range []string{"payer", "card", "رمز", "شماره کارت", "sql", "merchant_id"} { + if strings.Contains(lower, forbidden) { + return errors.New("avalai: unsafe content rejected") + } + } + } + if brief.Headline == "" || len(brief.EvidenceIDs) == 0 { + return errors.New("avalai: ungrounded brief") + } + for _, id := range brief.EvidenceIDs { + if _, ok := known[id]; !ok { + return errors.New("avalai: unknown evidence reference") + } } - if brief.Headline == "" || len(brief.EvidenceIDs) == 0 { return errors.New("avalai: ungrounded brief") } - for _, id := range brief.EvidenceIDs { if _, ok := known[id]; !ok { return errors.New("avalai: unknown evidence reference") } } } return nil } func growthTools() []map[string]any { - names := []string{"get_growth_context","get_event_evidence","get_peer_context","get_campaign_playbook"} + 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}}) } + 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}}) + } return tools } func aiCopySchema() map[string]any { briefProperties := map[string]any{} - for _, name := range []string{"headline","interpretation","offerSuggestion","messageSuggestion","preparationAction","limitation"} { briefProperties[name] = map[string]any{"type":"string"} } - briefProperties["evidenceIds"] = map[string]any{"type":"array","items":map[string]any{"type":"string"},"minItems":1} - return map[string]any{"type":"object","properties":map[string]any{"briefs":map[string]any{"type":"array","maxItems":5,"items":map[string]any{"type":"object","properties":briefProperties,"required":[]string{"headline","interpretation","offerSuggestion","messageSuggestion","preparationAction","limitation","evidenceIds"},"additionalProperties":false}}},"required":[]string{"briefs"},"additionalProperties":false} + for _, name := range []string{"headline", "interpretation", "offerSuggestion", "messageSuggestion", "preparationAction", "limitation"} { + briefProperties[name] = map[string]any{"type": "string"} + } + briefProperties["evidenceIds"] = map[string]any{"type": "array", "items": map[string]any{"type": "string"}, "minItems": 1} + return map[string]any{"type": "object", "properties": map[string]any{"briefs": map[string]any{"type": "array", "maxItems": 5, "items": map[string]any{"type": "object", "properties": briefProperties, "required": []string{"headline", "interpretation", "offerSuggestion", "messageSuggestion", "preparationAction", "limitation", "evidenceIds"}, "additionalProperties": false}}}, "required": []string{"briefs"}, "additionalProperties": false} } - diff --git a/internal/growthreport/avalai_test.go b/internal/growthreport/avalai_test.go index e4f2023..6812522 100644 --- a/internal/growthreport/avalai_test.go +++ b/internal/growthreport/avalai_test.go @@ -25,7 +25,9 @@ func TestValidateAICopyRejectsNumbersAndUnknownEvidence(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { err := ValidateAICopy(tt.copy, known) - if (err == nil) != tt.want { t.Fatalf("ValidateAICopy() error = %v, valid want %v", err, tt.want) } + if (err == nil) != tt.want { + t.Fatalf("ValidateAICopy() error = %v, valid want %v", err, tt.want) + } }) } } @@ -34,8 +36,12 @@ func TestAvalAIClientUsesBoundedStatelessRequest(t *testing.T) { t.Parallel() var request map[string]any server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if r.Header.Get("Authorization") != "Bearer test-key" { t.Errorf("authorization header missing") } - if err := json.NewDecoder(r.Body).Decode(&request); err != nil { t.Fatal(err) } + if r.Header.Get("Authorization") != "Bearer test-key" { + t.Errorf("authorization header missing") + } + if err := json.NewDecoder(r.Body).Decode(&request); err != nil { + t.Fatal(err) + } w.Header().Set("Content-Type", "application/json") _, _ = w.Write([]byte(`{"status":"completed","output":[{"type":"message","content":[{"type":"output_text","text":"{\"briefs\":[{\"headline\":\"فرصت مناسبی پیش رو دارید\",\"interpretation\":\"با الگوی موفق گذشته هم‌راستا است.\",\"offerSuggestion\":\"پیشنهاد مرتبط آماده کنید.\",\"messageSuggestion\":\"پیام کوتاه و روشن ارسال کنید.\",\"preparationAction\":\"ظرفیت پرداخت را بررسی کنید.\",\"limitation\":\"این پیشنهاد قطعی نیست.\",\"evidenceIds\":[\"event_teacher\"]}]}"}]}],"usage":{"input_tokens":10,"output_tokens":20}}`)) })) @@ -43,13 +49,55 @@ func TestAvalAIClientUsesBoundedStatelessRequest(t *testing.T) { client := NewAvalAIClient(AvalAIConfig{BaseURL: server.URL, APIKey: "test-key", Model: "test-model", Timeout: time.Second}, server.Client()) copy, usage, err := client.Generate(context.Background(), json.RawMessage(`{"evidenceIds":["event_teacher"]}`), map[string]struct{}{"event_teacher": {}}) - if err != nil { t.Fatal(err) } - if len(copy.Briefs) != 1 || usage.InputTokens != 10 || usage.OutputTokens != 20 { t.Fatalf("Generate() = %+v, %+v", copy, usage) } - if request["store"] != false || request["parallel_tool_calls"] != false { t.Fatalf("unsafe request controls: %#v", request) } - if request["max_output_tokens"] != float64(1600) { t.Fatalf("max_output_tokens = %#v", request["max_output_tokens"]) } + if err != nil { + t.Fatal(err) + } + if len(copy.Briefs) != 1 || usage.InputTokens != 10 || usage.OutputTokens != 20 { + t.Fatalf("Generate() = %+v, %+v", copy, usage) + } + if request["store"] != false || request["parallel_tool_calls"] != false { + t.Fatalf("unsafe request controls: %#v", request) + } + if request["max_output_tokens"] != float64(1600) { + t.Fatalf("max_output_tokens = %#v", request["max_output_tokens"]) + } tools, ok := request["tools"].([]any) - if !ok || len(tools) != 4 { t.Fatalf("tools = %#v", request["tools"]) } + if !ok || len(tools) != 4 { + t.Fatalf("tools = %#v", request["tools"]) + } encoded, _ := json.Marshal(request) - if strings.Contains(string(encoded), "merchantId") || strings.Contains(string(encoded), "payer") { t.Fatalf("sensitive field in request: %s", encoded) } + if strings.Contains(string(encoded), "merchantId") || strings.Contains(string(encoded), "payer") { + t.Fatalf("sensitive field in request: %s", encoded) + } } +type testTools struct{ calls int } + +func (t *testTools) Execute(_ context.Context, name string, _ json.RawMessage) (json.RawMessage, error) { + t.calls++ + return json.RawMessage(`{"state":"READY"}`), nil +} + +func TestAvalAIClientExecutesAllowlistedToolThenStopsAfterSecondTurn(t *testing.T) { + t.Parallel() + turn := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + turn++ + w.Header().Set("Content-Type", "application/json") + if turn == 1 { + _, _ = w.Write([]byte(`{"status":"completed","output":[{"type":"function_call","call_id":"call_1","name":"get_growth_context","arguments":"{\"evidenceId\":\"event_teacher\"}"}],"usage":{"input_tokens":2,"output_tokens":1}}`)) + return + } + _, _ = w.Write([]byte(`{"status":"completed","output":[{"type":"message","content":[{"type":"output_text","text":"{\"briefs\":[{\"headline\":\"فرصت مناسب\",\"interpretation\":\"هم‌راستا با شواهد است.\",\"offerSuggestion\":\"پیشنهاد مرتبط آماده کنید.\",\"messageSuggestion\":\"پیام روشن ارسال کنید.\",\"preparationAction\":\"مسیر پرداخت را بررسی کنید.\",\"limitation\":\"نتیجه قطعی نیست.\",\"evidenceIds\":[\"event_teacher\"]}]}"}]}],"usage":{"input_tokens":3,"output_tokens":4}}`)) + })) + defer server.Close() + executor := &testTools{} + client := NewAvalAIClient(AvalAIConfig{BaseURL: server.URL, APIKey: "key", Model: "model", Timeout: time.Second}, server.Client()) + _, usage, err := client.GenerateWithTools(context.Background(), json.RawMessage(`{}`), map[string]struct{}{"event_teacher": {}}, executor) + if err != nil { + t.Fatal(err) + } + if executor.calls != 1 || turn != 2 || usage.InputTokens != 5 || usage.OutputTokens != 5 { + t.Fatalf("calls=%d turns=%d usage=%+v", executor.calls, turn, usage) + } +} diff --git a/internal/growthreport/coordinator.go b/internal/growthreport/coordinator.go new file mode 100644 index 0000000..abaaf35 --- /dev/null +++ b/internal/growthreport/coordinator.go @@ -0,0 +1,102 @@ +package growthreport + +import ( + "context" + "errors" + "fmt" + "sync" + "time" + + "github.com/redis/go-redis/v9" +) + +var ErrRateLimited = errors.New("growth report: rate limited") +var ErrBudgetExhausted = errors.New("growth report: budget exhausted") + +type CoordinatorConfig struct { + CreatePerMinute, UncachedPerDay, DailyRequests, DailyTokens int64 + MaxConcurrent int +} +type Coordinator struct { + redis *redis.Client + config CoordinatorConfig + mu sync.Mutex + counters map[string]int64 + expiries map[string]time.Time + semaphore chan struct{} +} + +func NewCoordinator(client *redis.Client, config CoordinatorConfig) *Coordinator { + if config.MaxConcurrent < 1 { + config.MaxConcurrent = 4 + } + return &Coordinator{redis: client, config: config, counters: map[string]int64{}, expiries: map[string]time.Time{}, semaphore: make(chan struct{}, config.MaxConcurrent)} +} +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) ReserveBudget(ctx context.Context, tokens int64, now time.Time) error { + 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 { + return ErrBudgetExhausted + } + if err := c.incrementBy(ctx, "growth:budget:tok:"+tehranDate(now), tokens, c.config.DailyTokens, 36*time.Hour, now); err != nil { + return ErrBudgetExhausted + } + return nil +} +func (c *Coordinator) Enter(ctx context.Context) (func(), error) { + select { + case c.semaphore <- struct{}{}: + return func() { <-c.semaphore }, nil + case <-ctx.Done(): + return nil, ctx.Err() + } +} +func (c *Coordinator) Publish(ctx context.Context, reportID string) { + if c.redis != nil { + _ = 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) +} +func (c *Coordinator) incrementBy(ctx context.Context, key string, amount, limit int64, ttl time.Duration, now time.Time) error { + if limit < 1 { + return ErrRateLimited + } + if c.redis != nil { + 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 { + if value == 0 { + return ErrRateLimited + } + return nil + } + } + c.mu.Lock() + defer c.mu.Unlock() + if expiry := c.expiries[key]; !expiry.After(now) { + c.counters[key] = 0 + c.expiries[key] = now.Add(ttl) + } + next := c.counters[key] + amount + if next > limit { + return ErrRateLimited + } + c.counters[key] = next + return nil +} +func tehranDate(now time.Time) string { + location, err := time.LoadLocation("Asia/Tehran") + if err != nil { + location = time.FixedZone("Asia/Tehran", 12600) + } + return now.In(location).Format(time.DateOnly) +} diff --git a/internal/growthreport/coordinator_test.go b/internal/growthreport/coordinator_test.go new file mode 100644 index 0000000..95ac06e --- /dev/null +++ b/internal/growthreport/coordinator_test.go @@ -0,0 +1,29 @@ +package growthreport + +import ( + "context" + "errors" + "testing" + "time" +) + +func TestCoordinatorFallsBackToBoundedProcessCounters(t *testing.T) { + t.Parallel() + coordinator := NewCoordinator(nil, CoordinatorConfig{CreatePerMinute: 2, UncachedPerDay: 1, DailyRequests: 1, DailyTokens: 1600, MaxConcurrent: 1}) + now := time.Date(2026, 8, 21, 0, 0, 0, 0, time.UTC) + if err := coordinator.AllowCreate(context.Background(), 7, now); err != nil { + t.Fatal(err) + } + if err := coordinator.AllowCreate(context.Background(), 7, now); err != nil { + t.Fatal(err) + } + 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) + } +} diff --git a/internal/growthreport/model.go b/internal/growthreport/model.go index 23a7747..4f8c48d 100644 --- a/internal/growthreport/model.go +++ b/internal/growthreport/model.go @@ -8,17 +8,19 @@ import ( type Status string const ( - StatusQueued Status = "QUEUED" - StatusAnalyzing Status = "ANALYZING" + StatusQueued Status = "QUEUED" + StatusAnalyzing Status = "ANALYZING" StatusGenerating Status = "GENERATING" - StatusReady Status = "READY" - StatusPartial Status = "PARTIAL" - StatusFailed Status = "FAILED" + StatusReady Status = "READY" + StatusPartial Status = "PARTIAL" + StatusFailed Status = "FAILED" ) func (s Status) Terminal() bool { return s == StatusReady || s == StatusPartial || s == StatusFailed } -type Scope struct { TerminalID *string `json:"terminalId"` } +type Scope struct { + TerminalID *string `json:"terminalId"` +} type CampaignBrief struct { Headline string `json:"headline"` @@ -30,49 +32,50 @@ type CampaignBrief struct { EvidenceIDs []string `json:"evidenceIds"` } -type AICopy struct { Briefs []CampaignBrief `json:"briefs"` } +type AICopy struct { + Briefs []CampaignBrief `json:"briefs"` +} type Usage struct { - InputTokens int64 `json:"inputTokens"` + InputTokens int64 `json:"inputTokens"` OutputTokens int64 `json:"outputTokens"` - LatencyMS int64 `json:"latencyMs"` + LatencyMS int64 `json:"latencyMs"` } type Report struct { - ReportID string `json:"reportId"` - Status Status `json:"status"` - Scope Scope `json:"scope"` - HorizonDays int `json:"horizonDays"` - ForecastStartDate *time.Time `json:"forecastStartDate,omitempty"` + ReportID string `json:"reportId"` + Status Status `json:"status"` + Scope Scope `json:"scope"` + HorizonDays int `json:"horizonDays"` + ForecastStartDate *time.Time `json:"forecastStartDate,omitempty"` DeterministicEvidence json.RawMessage `json:"deterministicEvidence"` - AICopy *AICopy `json:"aiCopy,omitempty"` - FailureCode *string `json:"failureCode,omitempty"` - Cached bool `json:"cached"` - Stale bool `json:"stale"` - NextRefreshAt time.Time `json:"nextRefreshAt"` - CreatedAt time.Time `json:"createdAt"` - UpdatedAt time.Time `json:"updatedAt"` - EventsURL string `json:"eventsUrl"` + AICopy *AICopy `json:"aiCopy,omitempty"` + FailureCode *string `json:"failureCode,omitempty"` + Cached bool `json:"cached"` + Stale bool `json:"stale"` + NextRefreshAt time.Time `json:"nextRefreshAt"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` + EventsURL string `json:"eventsUrl"` } type Record struct { Report - MerchantID int64 - TerminalDBID *int64 - Fingerprint string + MerchantID int64 + TerminalDBID *int64 + Fingerprint string DatasetWatermark string - CatalogVersion string + CatalogVersion string AnalyticsVersion string - PromptVersion string - SchemaVersion string - Model string - Usage Usage - ManualRefresh bool + PromptVersion string + SchemaVersion string + Model string + Usage Usage + ManualRefresh bool TehranRefreshDay *time.Time } type CreateRequest struct { TerminalID string `json:"terminalId"` - Refresh bool `json:"refresh"` + Refresh bool `json:"refresh"` } - diff --git a/internal/growthreport/orchestrator.go b/internal/growthreport/orchestrator.go new file mode 100644 index 0000000..406c276 --- /dev/null +++ b/internal/growthreport/orchestrator.go @@ -0,0 +1,145 @@ +package growthreport + +import ( + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "time" + + "git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/growth" +) + +var ErrDisabled = errors.New("growth report: disabled") + +type ServiceConfig struct { + Enabled bool + Model, PromptVersion, SchemaVersion string +} +type Service struct { + repository *Repository + coordinator *Coordinator + config ServiceConfig + now func() time.Time +} + +func NewService(repository *Repository, coordinator *Coordinator, config ServiceConfig) *Service { + return &Service{repository: repository, coordinator: coordinator, config: config, now: time.Now} +} + +func (s *Service) Create(ctx context.Context, userID, merchantID int64, request CreateRequest, idempotencyKey string) (Record, bool, error) { + if !s.config.Enabled { + return Record{}, false, ErrDisabled + } + now := s.now().UTC() + 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 + } + 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}) + var refreshDay *time.Time + if request.Refresh { + day := tehranMidnight(now) + 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)}) + if err != nil { + return Record{}, false, err + } + if !cached { + if err := s.coordinator.AllowUncached(ctx, merchantID, now); err != nil { + _ = s.repository.Complete(ctx, record.ReportID, StatusFailed, json.RawMessage(`{}`), nil, Usage{}, "UNCACHED_SCOPE_LIMIT") + return Record{}, false, err + } + s.coordinator.Publish(ctx, record.ReportID) + } + record.Cached = cached + record.NextRefreshAt, err = s.repository.NextRefreshAt(ctx, merchantID, now) + return record, cached, err +} +func (s *Service) Get(ctx context.Context, merchantID int64, id string) (Record, error) { + record, err := s.repository.Get(ctx, merchantID, id) + if err == nil { + record.NextRefreshAt, _ = s.repository.NextRefreshAt(ctx, merchantID, s.now()) + } + return record, err +} +func (s *Service) Latest(ctx context.Context, merchantID int64, terminalID string) (Record, error) { + record, err := s.repository.Latest(ctx, merchantID, terminalID) + if err == nil { + record.Cached = true + record.NextRefreshAt, _ = s.repository.NextRefreshAt(ctx, merchantID, s.now()) + } + return record, err +} + +type Processor struct { + repository *Repository + builder Builder + avalai *AvalAIClient + coordinator *Coordinator + aiEnabled bool + now func() time.Time +} + +func NewProcessor(repository *Repository, builder Builder, avalai *AvalAIClient, coordinator *Coordinator, aiEnabled bool) *Processor { + return &Processor{repository: repository, builder: builder, avalai: avalai, coordinator: coordinator, aiEnabled: aiEnabled, now: time.Now} +} +func (p *Processor) ProcessNext(ctx context.Context) error { + record, err := p.repository.Claim(ctx) + if err != nil { + return err + } + terminal := "" + if record.Scope.TerminalID != nil { + terminal = *record.Scope.TerminalID + } + analysis, err := p.builder.Build(ctx, record.MerchantID, terminal) + if err != nil { + _ = p.repository.Complete(ctx, record.ReportID, StatusFailed, json.RawMessage(`{}`), nil, Usage{}, "ANALYTICS_FAILED") + return err + } + if err := p.repository.StoreEvidence(ctx, record.ReportID, analysis.Encoded); err != nil { + return err + } + copy := analysis.Fallback + status := StatusPartial + failure := "AI_SKIPPED" + if analysis.ShouldCallAI && p.aiEnabled && p.avalai != nil { + if err := p.coordinator.ReserveBudget(ctx, 1600, 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 { + return p.repository.Complete(ctx, record.ReportID, StatusReady, analysis.Encoded, &generated, usage, "") + } + failure = "AI_REJECTED" + } + } else { + failure = "AI_BUDGET_EXHAUSTED" + } + } + return p.repository.Complete(ctx, record.ReportID, status, analysis.Encoded, ©, Usage{}, failure) +} + +func newReportID() string { + data := make([]byte, 12) + if _, err := rand.Read(data); err != nil { + return fmt.Sprintf("growth_%d", time.Now().UnixNano()) + } + return "growth_" + hex.EncodeToString(data) +} +func tehranMidnight(now time.Time) time.Time { + location, err := time.LoadLocation("Asia/Tehran") + if err != nil { + location = time.FixedZone("Asia/Tehran", 12600) + } + local := now.In(location) + return time.Date(local.Year(), local.Month(), local.Day(), 0, 0, 0, 0, location) +} diff --git a/internal/growthreport/repository.go b/internal/growthreport/repository.go index 594c552..87f0e12 100644 --- a/internal/growthreport/repository.go +++ b/internal/growthreport/repository.go @@ -12,134 +12,249 @@ import ( ) var ( - ErrNotFound = errors.New("growth report: not found") - ErrUnknownTerminal = errors.New("growth report: unknown terminal") + ErrNotFound = errors.New("growth report: not found") + ErrUnknownTerminal = errors.New("growth report: unknown terminal") ErrIdempotencyConflict = errors.New("growth report: idempotency conflict") - ErrRefreshLimited = errors.New("growth report: refresh limited") + ErrRefreshLimited = errors.New("growth report: refresh limited") ) -type Repository struct { pool *pgxpool.Pool } +type Repository struct{ pool *pgxpool.Pool } func NewRepository(pool *pgxpool.Pool) *Repository { return &Repository{pool: pool} } type CreateRecord struct { - ID string - MerchantID int64 - TerminalID string - Fingerprint string - DatasetWatermark string + ID string + MerchantID int64 + TerminalID string + Fingerprint string + DatasetWatermark string ForecastStartDate *time.Time - CatalogVersion string - AnalyticsVersion string - PromptVersion string - SchemaVersion string - Model string - ManualRefresh bool - RefreshDay *time.Time - IdempotencyKey string - RequestHash string + CatalogVersion string + AnalyticsVersion string + PromptVersion string + SchemaVersion string + Model string + ManualRefresh bool + RefreshDay *time.Time + IdempotencyKey string + RequestHash string } func (r *Repository) CreateOrReuse(ctx context.Context, in CreateRecord) (Record, bool, error) { tx, err := r.pool.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.Serializable}) - if err != nil { return Record{}, false, fmt.Errorf("beginning report transaction: %w", err) } - defer func(){ _ = tx.Rollback(ctx) }() + if err != nil { + return Record{}, false, fmt.Errorf("beginning report transaction: %w", err) + } + defer func() { _ = tx.Rollback(ctx) }() + lockKey := fmt.Sprintf("growth:%d:%s:%t", in.MerchantID, in.Fingerprint, in.ManualRefresh) + if in.ManualRefresh && in.RefreshDay != nil { + lockKey = fmt.Sprintf("growth-refresh:%d:%s", in.MerchantID, in.RefreshDay.Format(time.DateOnly)) + } + if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,0))`, lockKey); err != nil { + return Record{}, false, fmt.Errorf("locking growth report request: %w", err) + } var terminalDBID *int64 if in.TerminalID != "" { var id int64 - if err := tx.QueryRow(ctx, `SELECT id FROM terminals WHERE merchant_id=$1 AND source_key=$2`, in.MerchantID, in.TerminalID).Scan(&id); errors.Is(err, pgx.ErrNoRows) { return Record{}, false, ErrUnknownTerminal } else if err != nil { return Record{}, false, fmt.Errorf("validating report terminal: %w", err) } + if err := tx.QueryRow(ctx, `SELECT id FROM terminals WHERE merchant_id=$1 AND source_key=$2`, in.MerchantID, in.TerminalID).Scan(&id); errors.Is(err, pgx.ErrNoRows) { + return Record{}, false, ErrUnknownTerminal + } else if err != nil { + return Record{}, false, fmt.Errorf("validating report terminal: %w", err) + } terminalDBID = &id } if in.IdempotencyKey != "" { var requestHash, reportID string err := tx.QueryRow(ctx, `SELECT request_hash,report_id FROM growth_report_idempotency WHERE merchant_id=$1 AND idempotency_key=$2`, in.MerchantID, in.IdempotencyKey).Scan(&requestHash, &reportID) if err == nil { - if requestHash != in.RequestHash { return Record{}, false, ErrIdempotencyConflict } + if requestHash != in.RequestHash { + return Record{}, false, ErrIdempotencyConflict + } record, err := getWith(ctx, tx, in.MerchantID, reportID) - if err != nil { return Record{}, false, err } + if err != nil { + return Record{}, false, err + } return record, true, nil } - if !errors.Is(err, pgx.ErrNoRows) { return Record{}, false, fmt.Errorf("reading report idempotency: %w", err) } + if !errors.Is(err, pgx.ErrNoRows) { + return Record{}, false, fmt.Errorf("reading report idempotency: %w", err) + } } if !in.ManualRefresh { record, err := currentWith(ctx, tx, in.MerchantID, in.Fingerprint) - if err == nil { return record, true, nil } - if !errors.Is(err, ErrNotFound) { return Record{}, false, err } + if err == nil { + return record, true, nil + } + if !errors.Is(err, ErrNotFound) { + return Record{}, false, err + } + } + if in.ManualRefresh && in.RefreshDay != nil { + var exists bool + if err := tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM growth_reports WHERE merchant_id=$1 AND manual_refresh=true AND tehran_refresh_day=$2)`, in.MerchantID, in.RefreshDay).Scan(&exists); err != nil { + return Record{}, false, fmt.Errorf("checking daily refresh: %w", err) + } + if exists { + return Record{}, false, ErrRefreshLimited + } } _, err = tx.Exec(ctx, `INSERT INTO growth_reports(id,merchant_id,terminal_id,terminal_source_key,fingerprint,status,forecast_start_date,dataset_watermark,catalog_version,analytics_version,prompt_version,schema_version,model,manual_refresh,tehran_refresh_day) VALUES($1,$2,$3,$4,$5,'QUEUED',$6,$7,$8,$9,$10,$11,$12,$13,$14)`, in.ID, in.MerchantID, terminalDBID, nilIfEmpty(in.TerminalID), in.Fingerprint, in.ForecastStartDate, in.DatasetWatermark, in.CatalogVersion, in.AnalyticsVersion, in.PromptVersion, in.SchemaVersion, in.Model, in.ManualRefresh, in.RefreshDay) if err != nil { - if in.ManualRefresh && isUniqueViolation(err) { return Record{}, false, ErrRefreshLimited } - if !in.ManualRefresh && isUniqueViolation(err) { - record, lookupErr := currentWith(ctx, tx, in.MerchantID, in.Fingerprint) - if lookupErr == nil { return record, true, nil } + if in.ManualRefresh && isUniqueViolation(err) { + return Record{}, false, ErrRefreshLimited } return Record{}, false, fmt.Errorf("creating growth report: %w", err) } if in.IdempotencyKey != "" { - if _, err := tx.Exec(ctx, `INSERT INTO growth_report_idempotency(merchant_id,idempotency_key,request_hash,report_id) VALUES($1,$2,$3,$4)`, in.MerchantID, in.IdempotencyKey, in.RequestHash, in.ID); err != nil { return Record{}, false, fmt.Errorf("creating report idempotency: %w", err) } + if _, err := tx.Exec(ctx, `INSERT INTO growth_report_idempotency(merchant_id,idempotency_key,request_hash,report_id) VALUES($1,$2,$3,$4)`, in.MerchantID, in.IdempotencyKey, in.RequestHash, in.ID); err != nil { + return Record{}, false, fmt.Errorf("creating report idempotency: %w", err) + } + } + record, err := getWith(ctx, tx, in.MerchantID, in.ID) + if err != nil { + return Record{}, false, err + } + if err := tx.Commit(ctx); err != nil { + return Record{}, false, fmt.Errorf("committing growth report: %w", err) } - record, err := getWith(ctx, tx, in.MerchantID, in.ID); if err != nil { return Record{}, false, err } - if err := tx.Commit(ctx); err != nil { return Record{}, false, fmt.Errorf("committing growth report: %w", err) } return record, false, nil } -func (r *Repository) Get(ctx context.Context, merchantID int64, id string) (Record, error) { return getWith(ctx, r.pool, merchantID, id) } +func (r *Repository) Get(ctx context.Context, merchantID int64, id string) (Record, error) { + return getWith(ctx, r.pool, merchantID, id) +} func (r *Repository) Latest(ctx context.Context, merchantID int64, terminalID string) (Record, error) { return scanRecord(r.pool.QueryRow(ctx, reportSelect+` WHERE merchant_id=$1 AND terminal_source_key IS NOT DISTINCT FROM $2::text AND status IN ('READY','PARTIAL') ORDER BY created_at DESC LIMIT 1`, merchantID, nilIfEmpty(terminalID))) } func (r *Repository) Claim(ctx context.Context) (Record, error) { - tx, err := r.pool.Begin(ctx); if err != nil { return Record{}, fmt.Errorf("beginning report claim: %w", err) } - defer func(){ _ = tx.Rollback(ctx) }() + tx, err := r.pool.Begin(ctx) + if err != nil { + return Record{}, fmt.Errorf("beginning report claim: %w", err) + } + defer func() { _ = tx.Rollback(ctx) }() row := tx.QueryRow(ctx, reportSelect+` WHERE id=(SELECT id FROM growth_reports WHERE status='QUEUED' ORDER BY created_at FOR UPDATE SKIP LOCKED LIMIT 1)`) record, err := scanRecord(row) - if errors.Is(err, ErrNotFound) { return Record{}, ErrNotFound } - if err != nil { return Record{}, err } - if _, err := tx.Exec(ctx, `UPDATE growth_reports SET status='ANALYZING',updated_at=now() WHERE id=$1`, record.ReportID); err != nil { return Record{}, fmt.Errorf("claiming growth report: %w", err) } - record.Status = StatusAnalyzing; record.UpdatedAt = time.Now().UTC() - if err := tx.Commit(ctx); err != nil { return Record{}, fmt.Errorf("committing report claim: %w", err) } + if errors.Is(err, ErrNotFound) { + return Record{}, ErrNotFound + } + if err != nil { + return Record{}, err + } + if _, err := tx.Exec(ctx, `UPDATE growth_reports SET status='ANALYZING',updated_at=now() WHERE id=$1`, record.ReportID); err != nil { + return Record{}, fmt.Errorf("claiming growth report: %w", err) + } + record.Status = StatusAnalyzing + record.UpdatedAt = time.Now().UTC() + if err := tx.Commit(ctx); err != nil { + return Record{}, fmt.Errorf("committing report claim: %w", err) + } return record, nil } func (r *Repository) StoreEvidence(ctx context.Context, reportID string, evidence json.RawMessage) error { command, err := r.pool.Exec(ctx, `UPDATE growth_reports SET deterministic_evidence=$2,status='GENERATING',updated_at=now() WHERE id=$1 AND status='ANALYZING'`, reportID, evidence) - if err != nil { return fmt.Errorf("storing deterministic evidence: %w", err) } - if command.RowsAffected()!=1 { return ErrNotFound }; return nil + if err != nil { + return fmt.Errorf("storing deterministic evidence: %w", err) + } + if command.RowsAffected() != 1 { + return ErrNotFound + } + return nil } func (r *Repository) Complete(ctx context.Context, reportID string, status Status, evidence json.RawMessage, copy *AICopy, usage Usage, failureCode string) error { - if status != StatusReady && status != StatusPartial && status != StatusFailed { return errors.New("growth report: invalid terminal status") } + if status != StatusReady && status != StatusPartial && status != StatusFailed { + return errors.New("growth report: invalid terminal status") + } var ai any - if copy != nil { ai = copy } + if copy != nil { + ai = copy + } _, err := r.pool.Exec(ctx, `UPDATE growth_reports SET status=$2,deterministic_evidence=$3,ai_response=$4,failure_code=$5,input_tokens=$6,output_tokens=$7,latency_ms=$8,updated_at=now(),completed_at=now() WHERE id=$1`, reportID, status, evidence, ai, nilIfEmpty(failureCode), usage.InputTokens, usage.OutputTokens, usage.LatencyMS) - if err != nil { return fmt.Errorf("completing growth report: %w", err) }; return nil + if err != nil { + return fmt.Errorf("completing growth report: %w", err) + } + return nil } const reportSelect = `SELECT id,merchant_id,terminal_id,terminal_source_key,fingerprint,status,horizon_days,forecast_start_date,dataset_watermark,catalog_version,analytics_version,prompt_version,schema_version,model,deterministic_evidence,ai_response,failure_code,input_tokens,output_tokens,latency_ms,manual_refresh,tehran_refresh_day,created_at,updated_at FROM growth_reports` -type rowScanner interface { Scan(...any) error } -type queryer interface { QueryRow(context.Context, string, ...any) pgx.Row } - -func getWith(ctx context.Context, q queryer, merchantID int64, id string) (Record,error) { return scanRecord(q.QueryRow(ctx, reportSelect+` WHERE merchant_id=$1 AND id=$2`, merchantID,id)) } -func currentWith(ctx context.Context, q queryer, merchantID int64, fingerprint string) (Record,error) { return scanRecord(q.QueryRow(ctx, reportSelect+` WHERE merchant_id=$1 AND fingerprint=$2 AND manual_refresh=false AND status IN ('QUEUED','ANALYZING','GENERATING','READY','PARTIAL') ORDER BY created_at DESC LIMIT 1`,merchantID,fingerprint)) } - -func scanRecord(row rowScanner) (Record,error) { - var r Record; var status string; var evidence []byte; var ai []byte; var terminalSource *string - err := row.Scan(&r.ReportID,&r.MerchantID,&r.TerminalDBID,&terminalSource,&r.Fingerprint,&status,&r.HorizonDays,&r.ForecastStartDate,&r.DatasetWatermark,&r.CatalogVersion,&r.AnalyticsVersion,&r.PromptVersion,&r.SchemaVersion,&r.Model,&evidence,&ai,&r.FailureCode,&r.Usage.InputTokens,&r.Usage.OutputTokens,&r.Usage.LatencyMS,&r.ManualRefresh,&r.TehranRefreshDay,&r.CreatedAt,&r.UpdatedAt) - if errors.Is(err,pgx.ErrNoRows){ return Record{},ErrNotFound }; if err!=nil{return Record{},fmt.Errorf("scanning growth report: %w",err)} - r.Status=Status(status); if terminalSource!=nil { r.Scope.TerminalID=terminalSource }; r.DeterministicEvidence=json.RawMessage(evidence); r.EventsURL="/api/v1/analytics/growth-reports/"+r.ReportID+"/events"; r.NextRefreshAt=NextTehranRefresh(r.CreatedAt) - if len(ai)>0 { var copy AICopy; if json.Unmarshal(ai,©)==nil { r.AICopy=© } } - return r,nil +type rowScanner interface{ Scan(...any) error } +type queryer interface { + QueryRow(context.Context, string, ...any) pgx.Row } -func nilIfEmpty(value string) any { if value=="" { return nil }; return value } +func getWith(ctx context.Context, q queryer, merchantID int64, id string) (Record, error) { + return scanRecord(q.QueryRow(ctx, reportSelect+` WHERE merchant_id=$1 AND id=$2`, merchantID, id)) +} +func currentWith(ctx context.Context, q queryer, merchantID int64, fingerprint string) (Record, error) { + return scanRecord(q.QueryRow(ctx, reportSelect+` WHERE merchant_id=$1 AND fingerprint=$2 AND manual_refresh=false AND status IN ('QUEUED','ANALYZING','GENERATING','READY','PARTIAL') ORDER BY created_at DESC LIMIT 1`, merchantID, fingerprint)) +} + +func scanRecord(row rowScanner) (Record, error) { + var r Record + var status string + var evidence []byte + var ai []byte + var terminalSource *string + err := row.Scan(&r.ReportID, &r.MerchantID, &r.TerminalDBID, &terminalSource, &r.Fingerprint, &status, &r.HorizonDays, &r.ForecastStartDate, &r.DatasetWatermark, &r.CatalogVersion, &r.AnalyticsVersion, &r.PromptVersion, &r.SchemaVersion, &r.Model, &evidence, &ai, &r.FailureCode, &r.Usage.InputTokens, &r.Usage.OutputTokens, &r.Usage.LatencyMS, &r.ManualRefresh, &r.TehranRefreshDay, &r.CreatedAt, &r.UpdatedAt) + if errors.Is(err, pgx.ErrNoRows) { + return Record{}, ErrNotFound + } + if err != nil { + return Record{}, fmt.Errorf("scanning growth report: %w", err) + } + r.Status = Status(status) + if terminalSource != nil { + r.Scope.TerminalID = terminalSource + } + r.DeterministicEvidence = json.RawMessage(evidence) + r.EventsURL = "/api/v1/analytics/growth-reports/" + r.ReportID + "/events" + if len(ai) > 0 { + var copy AICopy + if json.Unmarshal(ai, ©) == nil { + r.AICopy = © + } + } + return r, nil +} + +func nilIfEmpty(value string) any { + if value == "" { + return nil + } + return value +} func isUniqueViolation(err error) bool { - type sqlState interface { SQLState() string } + type sqlState interface{ SQLState() string } var state sqlState - return errors.As(err,&state) && state.SQLState()=="23505" + return errors.As(err, &state) && state.SQLState() == "23505" } +func (r *Repository) NextRefreshAt(ctx context.Context, merchantID int64, now time.Time) (time.Time, error) { + var last *time.Time + if err := r.pool.QueryRow(ctx, `SELECT max(created_at) FROM growth_reports WHERE merchant_id=$1 AND manual_refresh=true`, merchantID).Scan(&last); err != nil { + return time.Time{}, fmt.Errorf("loading refresh quota: %w", err) + } + if last == nil { + return now.UTC(), nil + } + return NextTehranRefresh(*last), nil +} + +func (r *Repository) DatasetMetadata(ctx context.Context) (string, time.Time, error) { + var sha string + var latest time.Time + err := r.pool.QueryRow(ctx, `SELECT ir.source_sha256,max(ps.created_at AT TIME ZONE 'Asia/Tehran')::date + FROM payment_sessions ps JOIN import_runs ir ON ir.id=ps.import_run_id WHERE ir.status='COMPLETED' + GROUP BY ir.id,ir.source_sha256 ORDER BY ir.completed_at DESC LIMIT 1`).Scan(&sha, &latest) + if err != nil { + return "", time.Time{}, fmt.Errorf("loading growth dataset metadata: %w", err) + } + return sha + ":" + latest.Format(time.DateOnly), latest.AddDate(0, 0, 1), nil +} diff --git a/internal/growthreport/service.go b/internal/growthreport/service.go index 1986baa..93d53a0 100644 --- a/internal/growthreport/service.go +++ b/internal/growthreport/service.go @@ -10,4 +10,3 @@ func RequestHash(merchantID int64, terminalID string, refresh bool) string { sum := sha256.Sum256([]byte(fmt.Sprintf("%d\x00%s\x00%t", merchantID, terminalID, refresh))) return hex.EncodeToString(sum[:]) } - diff --git a/internal/growthreport/service_test.go b/internal/growthreport/service_test.go index 5b7783c..2ee7eff 100644 --- a/internal/growthreport/service_test.go +++ b/internal/growthreport/service_test.go @@ -8,8 +8,12 @@ import ( func TestRequestHashIsStableAndScopeSensitive(t *testing.T) { t.Parallel() a := RequestHash(7, "T-1", false) - if a != RequestHash(7, "T-1", false) { t.Fatal("RequestHash() is not stable") } - if a == RequestHash(7, "T-2", false) || a == RequestHash(7, "T-1", true) { t.Fatal("RequestHash() ignored request scope") } + if a != RequestHash(7, "T-1", false) { + t.Fatal("RequestHash() is not stable") + } + if a == RequestHash(7, "T-2", false) || a == RequestHash(7, "T-1", true) { + t.Fatal("RequestHash() ignored request scope") + } } func TestNextTehranRefreshUsesCalendarBoundary(t *testing.T) { @@ -18,6 +22,7 @@ func TestNextTehranRefreshUsesCalendarBoundary(t *testing.T) { next := NextTehranRefresh(now) location, _ := time.LoadLocation("Asia/Tehran") local := next.In(location) - if local.Hour() != 0 || local.Minute() != 0 || !next.After(now) { t.Fatalf("NextTehranRefresh() = %s", next) } + if local.Hour() != 0 || local.Minute() != 0 || !next.After(now) { + t.Fatalf("NextTehranRefresh() = %s", next) + } } - diff --git a/internal/growthreport/time.go b/internal/growthreport/time.go index 63fff96..fc13daf 100644 --- a/internal/growthreport/time.go +++ b/internal/growthreport/time.go @@ -4,8 +4,9 @@ import "time" func NextTehranRefresh(now time.Time) time.Time { location, err := time.LoadLocation("Asia/Tehran") - if err != nil { location = time.FixedZone("Asia/Tehran", 3*60*60+30*60) } + if err != nil { + location = time.FixedZone("Asia/Tehran", 3*60*60+30*60) + } local := now.In(location) return time.Date(local.Year(), local.Month(), local.Day()+1, 0, 0, 0, 0, location).UTC() } -