feat: add safe growth report persistence and AvalAI client
This commit is contained in:
110
internal/growthreport/avalai.go
Normal file
110
internal/growthreport/avalai.go
Normal file
@@ -0,0 +1,110 @@
|
||||
package growthreport
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
"unicode"
|
||||
)
|
||||
|
||||
const maxAvalAIResponseBytes = 1 << 20
|
||||
|
||||
type AvalAIConfig struct {
|
||||
BaseURL string
|
||||
APIKey string
|
||||
Model string
|
||||
Timeout time.Duration
|
||||
}
|
||||
|
||||
type AvalAIClient struct { config AvalAIConfig; client *http.Client }
|
||||
|
||||
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{} }
|
||||
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{
|
||||
"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()}},
|
||||
}
|
||||
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 }
|
||||
last = err
|
||||
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()
|
||||
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) }
|
||||
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 := 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) }
|
||||
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 } } } }
|
||||
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
|
||||
}
|
||||
|
||||
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") }
|
||||
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") } }
|
||||
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") } }
|
||||
}
|
||||
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"}
|
||||
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}}) }
|
||||
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}
|
||||
}
|
||||
|
||||
55
internal/growthreport/avalai_test.go
Normal file
55
internal/growthreport/avalai_test.go
Normal file
@@ -0,0 +1,55 @@
|
||||
package growthreport
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestValidateAICopyRejectsNumbersAndUnknownEvidence(t *testing.T) {
|
||||
t.Parallel()
|
||||
tests := []struct {
|
||||
name string
|
||||
copy AICopy
|
||||
want bool
|
||||
}{
|
||||
{name: "grounded Persian copy", copy: AICopy{Briefs: []CampaignBrief{{Headline: "فرصت مناسبی پیش رو دارید", Interpretation: "این مناسبت با الگوی موفق گذشته همراستا است.", EvidenceIDs: []string{"event_teacher"}}}}, want: true},
|
||||
{name: "numeric claim", copy: AICopy{Briefs: []CampaignBrief{{Headline: "رشد ۲۰ درصدی", EvidenceIDs: []string{"event_teacher"}}}}, want: false},
|
||||
{name: "unknown evidence", copy: AICopy{Briefs: []CampaignBrief{{Headline: "فرصت مناسب", EvidenceIDs: []string{"unknown"}}}}, want: false},
|
||||
}
|
||||
known := map[string]struct{}{"event_teacher": {}}
|
||||
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) }
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
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) }
|
||||
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}}`))
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
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"]) }
|
||||
tools, ok := request["tools"].([]any)
|
||||
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) }
|
||||
}
|
||||
|
||||
78
internal/growthreport/model.go
Normal file
78
internal/growthreport/model.go
Normal file
@@ -0,0 +1,78 @@
|
||||
package growthreport
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"time"
|
||||
)
|
||||
|
||||
type Status string
|
||||
|
||||
const (
|
||||
StatusQueued Status = "QUEUED"
|
||||
StatusAnalyzing Status = "ANALYZING"
|
||||
StatusGenerating Status = "GENERATING"
|
||||
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 CampaignBrief struct {
|
||||
Headline string `json:"headline"`
|
||||
Interpretation string `json:"interpretation"`
|
||||
OfferSuggestion string `json:"offerSuggestion"`
|
||||
MessageSuggestion string `json:"messageSuggestion"`
|
||||
PreparationAction string `json:"preparationAction"`
|
||||
Limitation string `json:"limitation"`
|
||||
EvidenceIDs []string `json:"evidenceIds"`
|
||||
}
|
||||
|
||||
type AICopy struct { Briefs []CampaignBrief `json:"briefs"` }
|
||||
|
||||
type Usage struct {
|
||||
InputTokens int64 `json:"inputTokens"`
|
||||
OutputTokens int64 `json:"outputTokens"`
|
||||
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"`
|
||||
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"`
|
||||
}
|
||||
|
||||
type Record struct {
|
||||
Report
|
||||
MerchantID int64
|
||||
TerminalDBID *int64
|
||||
Fingerprint string
|
||||
DatasetWatermark string
|
||||
CatalogVersion string
|
||||
AnalyticsVersion string
|
||||
PromptVersion string
|
||||
SchemaVersion string
|
||||
Model string
|
||||
Usage Usage
|
||||
ManualRefresh bool
|
||||
TehranRefreshDay *time.Time
|
||||
}
|
||||
|
||||
type CreateRequest struct {
|
||||
TerminalID string `json:"terminalId"`
|
||||
Refresh bool `json:"refresh"`
|
||||
}
|
||||
|
||||
145
internal/growthreport/repository.go
Normal file
145
internal/growthreport/repository.go
Normal file
@@ -0,0 +1,145 @@
|
||||
package growthreport
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
var (
|
||||
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")
|
||||
)
|
||||
|
||||
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
|
||||
ForecastStartDate *time.Time
|
||||
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) }()
|
||||
|
||||
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) }
|
||||
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 }
|
||||
record, err := getWith(ctx, tx, in.MerchantID, reportID)
|
||||
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 !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 }
|
||||
}
|
||||
_, 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 }
|
||||
}
|
||||
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) }
|
||||
}
|
||||
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) 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) }()
|
||||
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) }
|
||||
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
|
||||
}
|
||||
|
||||
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") }
|
||||
var ai any
|
||||
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
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
func nilIfEmpty(value string) any { if value=="" { return nil }; return value }
|
||||
|
||||
func isUniqueViolation(err error) bool {
|
||||
type sqlState interface { SQLState() string }
|
||||
var state sqlState
|
||||
return errors.As(err,&state) && state.SQLState()=="23505"
|
||||
}
|
||||
|
||||
13
internal/growthreport/service.go
Normal file
13
internal/growthreport/service.go
Normal file
@@ -0,0 +1,13 @@
|
||||
package growthreport
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
)
|
||||
|
||||
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[:])
|
||||
}
|
||||
|
||||
23
internal/growthreport/service_test.go
Normal file
23
internal/growthreport/service_test.go
Normal file
@@ -0,0 +1,23 @@
|
||||
package growthreport
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
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") }
|
||||
}
|
||||
|
||||
func TestNextTehranRefreshUsesCalendarBoundary(t *testing.T) {
|
||||
t.Parallel()
|
||||
now := time.Date(2026, 8, 21, 20, 0, 0, 0, time.UTC)
|
||||
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) }
|
||||
}
|
||||
|
||||
11
internal/growthreport/time.go
Normal file
11
internal/growthreport/time.go
Normal file
@@ -0,0 +1,11 @@
|
||||
package growthreport
|
||||
|
||||
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) }
|
||||
local := now.In(location)
|
||||
return time.Date(local.Year(), local.Month(), local.Day()+1, 0, 0, 0, 0, location).UTC()
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user