From e6afaa7a1043cad0bdcf05fa35136252486182e3 Mon Sep 17 00:00:00 2001 From: Amirhossein Khalili Date: Fri, 21 Aug 2026 02:57:46 +0330 Subject: [PATCH] fix: keep growth analysis available per terminal --- cmd/migrate/main.go | 4 +- docs/GROWTH-ANALYST-SPEC.md | 6 +- docs/openapi.yaml | 9 ++- internal/growth/analytics_test.go | 20 +++++ internal/growth/catalog_test.go | 6 +- internal/growth/forecast.go | 29 ++++--- internal/growth/types.go | 3 +- internal/growthreport/orchestrator.go | 14 ++-- internal/growthreport/repository.go | 41 +++++----- .../repository_integration_test.go | 80 +++++++++++++++++++ migrations/000003_ai_summary_refresh.sql | 13 +++ 11 files changed, 180 insertions(+), 45 deletions(-) create mode 100644 migrations/000003_ai_summary_refresh.sql diff --git a/cmd/migrate/main.go b/cmd/migrate/main.go index 61420b3..977e41a 100644 --- a/cmd/migrate/main.go +++ b/cmd/migrate/main.go @@ -25,7 +25,7 @@ func main() { os.Exit(1) } defer pool.Close() - for _, path := range []string{"migrations/000001_mvp.sql", "migrations/000002_growth_analyst.sql"} { + for _, path := range []string{"migrations/000001_mvp.sql", "migrations/000002_growth_analyst.sql", "migrations/000003_ai_summary_refresh.sql"} { migration, err := os.ReadFile(path) if err != nil { logger.Error("reading migration failed", "path", path, "error", err) @@ -36,5 +36,5 @@ func main() { os.Exit(1) } } - logger.Info("migration complete", "version", 2) + logger.Info("migration complete", "version", 3) } diff --git a/docs/GROWTH-ANALYST-SPEC.md b/docs/GROWTH-ANALYST-SPEC.md index 07e79da..8c70e05 100644 --- a/docs/GROWTH-ANALYST-SPEC.md +++ b/docs/GROWTH-ANALYST-SPEC.md @@ -13,7 +13,7 @@ Give an authenticated merchant a trustworthy 90-day Persian growth outlook: hist - Aggregate session count and paid Rial amount by Tehran-local day for the authenticated merchant and optional owned terminal. `Paid` and `Verified` are successful; money remains integer Rial at session grain. - Forecast begins the day after the latest complete imported dataset date and covers 90 days. Count and amount are modeled independently from weekday-aware rolling baselines, bounded trend adjustment, catalogued event uplift, and empirical residual P10/P50/P90 values. -- Rolling-origin backtests classify usable output as `HIGH` or `MEDIUM`; other results return `INSUFFICIENT_DATA` and do not call AvalAI. +- Rolling-origin backtests classify output as `HIGH`, `MEDIUM`, or `LOW`. Limited-history and poorly validated forecasts remain visible as `LOW` with wider uncertainty; only a scope with no historical observations returns `INSUFFICIENT_DATA`. - Historical event effect compares matching weekdays from the preceding eight non-event weeks. At least four observations and 100 baseline sessions are required. A material win/decline is at least 20%. - Upcoming similarity prefers an exact event family, then catalog tag similarity of at least 0.60. Return at most five opportunities and three risks. A risk requires a forecast at least 20% below a baseline with at least 100 sessions. - Merchant scope may include peer context only under the existing threshold of 10 other same-category merchants and 1,000 combined peer sessions, with one equally weighted value per merchant. Terminal scope always suppresses peers. @@ -25,7 +25,7 @@ Give an authenticated merchant a trustworthy 90-day Persian growth outlook: hist - The consumer-owned `GrowthAnalyticsTools` boundary exposes only `get_growth_context`, `get_event_evidence`, `get_peer_context`, and `get_campaign_playbook`. Scope is injected from the authenticated job, not accepted from model arguments. - Use `POST /v1/responses`, `store:false`, strict function schemas, strict structured output, `parallel_tool_calls:false`, no more than four tool calls and two model turns, 1,600 output tokens, a 30-second deadline, and one transient retry. - Treat model output as untrusted. Accept only known evidence ids and qualitative Persian fields. Reject numeric claims, unsafe language, unknown references, refusal/incomplete output, or schema drift and use deterministic Persian fallback copy. -- Skip AvalAI when no material opportunity/risk exists or confidence is insufficient. AI failure produces `PARTIAL`, never an empty report. +- Skip AvalAI when no material opportunity/risk exists or there are no historical observations. Low-confidence evidence may be interpreted, but the response must preserve its limitation. AI failure produces `PARTIAL`, never an empty report. ## Persistence, caching, jobs, and limits @@ -33,7 +33,7 @@ Give an authenticated merchant a trustworthy 90-day Persian growth outlook: hist - A fingerprint includes merchant/terminal scope, import watermark, forecast date/horizon, catalog, analytics, prompt/schema, and model versions. A matching successful report is reused without an AI call. - Keep the current matching report indefinitely and superseded/manual-refresh revisions for 90 days. Concurrent identical requests collapse to one job. - States are `QUEUED`, `ANALYZING`, `GENERATING`, `READY`, `PARTIAL`, and `FAILED`. Preserve the previous successful report during refresh/failure. -- Defaults: create 10/user/minute; five uncached scopes/merchant/Tehran day; one merchant-wide manual refresh/Tehran day; four concurrent generations globally; explicit production daily request/token budgets; one active generation per fingerprint. +- Defaults: create 10/user/minute; five uncached scopes/merchant/Tehran day; four concurrent generations globally; explicit production daily request/token budgets; one active generation per fingerprint. Manual retries remain available for a scope until it has a persisted `READY` AI summary; after that, one manual refresh per terminal scope is available per Tehran day. The all-terminals selection is a separate scope. - Budget exhaustion returns deterministic `PARTIAL` output. Only an unavailable manual refresh returns `429` with `nextRefreshAt`/`Retry-After`. - Structured logs include safe report/version/status/cache/usage/latency fields without prompts, transaction values, usernames, or sensitive identifiers. diff --git a/docs/openapi.yaml b/docs/openapi.yaml index 7d58641..e5e60ed 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -1010,7 +1010,7 @@ components: terminalId: {type: string, minLength: 1, maxLength: 128} refresh: type: boolean - description: When omitted, the request reuses the current matching report. + description: When omitted, the request reuses the current matching report. Manual retries remain available until the requested scope has persisted AI copy. GrowthRange: type: object additionalProperties: false @@ -1037,7 +1037,7 @@ components: properties: startDate: {type: string, format: date-time} horizonDays: {type: integer, enum: [90]} - confidence: {type: string, enum: [HIGH, MEDIUM, INSUFFICIENT_DATA]} + confidence: {type: string, enum: [HIGH, MEDIUM, LOW, INSUFFICIENT_DATA]} suppressionReason: {type: string} points: {type: array, maxItems: 90, items: {$ref: '#/components/schemas/GrowthForecastPoint'}} backtest: @@ -1167,7 +1167,10 @@ components: failureCode: {type: string} cached: {type: boolean} stale: {type: boolean} - nextRefreshAt: {type: string, format: date-time} + nextRefreshAt: + type: string + format: date-time + description: Current time while no AI summary exists for the scope; otherwise the next refresh boundary for that terminal scope in the Tehran day. The all-terminals selection is a separate scope. createdAt: {type: string, format: date-time} updatedAt: {type: string, format: date-time} eventsUrl: {type: string, format: uri-reference} diff --git a/internal/growth/analytics_test.go b/internal/growth/analytics_test.go index f566c8a..3f8f074 100644 --- a/internal/growth/analytics_test.go +++ b/internal/growth/analytics_test.go @@ -63,6 +63,26 @@ func TestBuildForecastReturnsOrderedNinetyDayIntegerRanges(t *testing.T) { } } +func TestBuildForecastUsesLimitedHistoryWithLowConfidence(t *testing.T) { + start := date("2026-07-01") + history := make([]DailyPoint, 0, 14) + for daysAgo := 14; daysAgo > 0; daysAgo-- { + day := start.AddDate(0, 0, -daysAgo) + value := int64(80 + int(day.Weekday())*5) + history = append(history, DailyPoint{Date: day, SessionCount: value, PaidSessionCount: value, PaidAmountRial: value * 100_000}) + } + + forecast := BuildForecast(history, nil, start, 90) + if forecast.Confidence != ConfidenceLow || len(forecast.Points) != 90 { + t.Fatalf("limited history should produce a low-confidence forecast: confidence=%s points=%d", forecast.Confidence, len(forecast.Points)) + } + for _, point := range forecast.Points { + if point.PaidSessions.P10 > point.PaidSessions.P50 || point.PaidSessions.P50 > point.PaidSessions.P90 { + t.Fatalf("limited-history range is unordered: %+v", point.PaidSessions) + } + } +} + func TestDetectRisksRequiresTwentyPercentDropAndBaselineSize(t *testing.T) { forecast := Forecast{Confidence: ConfidenceMedium, Points: []ForecastPoint{ {Date: date("2026-07-01"), BaselineSessions: 200, PaidSessions: Range{P50: 150}, BaselineAmountRial: 10_000_000, PaidAmountRial: Range{P50: 7_000_000}}, diff --git a/internal/growth/catalog_test.go b/internal/growth/catalog_test.go index fc4e57b..421c0bf 100644 --- a/internal/growth/catalog_test.go +++ b/internal/growth/catalog_test.go @@ -80,7 +80,7 @@ func TestAnalyzeEventEffectUsesCompleteComparableWindows(t *testing.T) { } } -func TestForecastSuppressesPoorBacktest(t *testing.T) { +func TestForecastKeepsPoorBacktestVisibleWithLowConfidence(t *testing.T) { start := date("2026-07-01") history := make([]DailyPoint, 0, 140) for i := 140; i > 0; i-- { @@ -91,8 +91,8 @@ func TestForecastSuppressesPoorBacktest(t *testing.T) { history = append(history, DailyPoint{Date: start.AddDate(0, 0, -i), SessionCount: value, PaidSessionCount: value, PaidAmountRial: value * 1_000}) } forecast := BuildForecast(history, nil, start, 90) - if forecast.Backtest.Origins < 4 || forecast.Confidence != ConfidenceInsufficientData || len(forecast.Points) != 0 { - t.Fatalf("poor model should be suppressed: %+v", forecast) + if forecast.Backtest.Origins < 4 || forecast.Confidence != ConfidenceLow || len(forecast.Points) != 90 { + t.Fatalf("poor model should remain visible with low confidence: %+v", forecast) } } diff --git a/internal/growth/forecast.go b/internal/growth/forecast.go index bce5747..42f743e 100644 --- a/internal/growth/forecast.go +++ b/internal/growth/forecast.go @@ -17,20 +17,17 @@ func BuildForecast(history []DailyPoint, catalog []Event, start time.Time, horiz } result := Forecast{StartDate: normalizeDate(start), HorizonDays: horizonDays, Confidence: ConfidenceInsufficientData, SuppressionReason: SuppressionInsufficientHistory, Points: []ForecastPoint{}} ordered := normalizedHistory(history, start) - if len(ordered) < minimumHistoryDays { + if len(ordered) == 0 { return result } result.Backtest = backtest(ordered, catalog) result.Confidence = classifyConfidence(result.Backtest) - if result.Confidence == ConfidenceInsufficientData { - return result - } result.SuppressionReason = "" - result.Points = forecastPoints(ordered, catalog, result.StartDate, horizonDays) + result.Points = forecastPoints(ordered, catalog, result.StartDate, horizonDays, result.Confidence) return result } -func forecastPoints(history []DailyPoint, catalog []Event, start time.Time, horizon int) []ForecastPoint { +func forecastPoints(history []DailyPoint, catalog []Event, start time.Time, horizon int, confidence Confidence) []ForecastPoint { sessionResiduals, amountResiduals := modelResiduals(history, catalog) points := make([]ForecastPoint, 0, horizon) for index := 0; index < horizon; index++ { @@ -38,8 +35,8 @@ func forecastPoints(history []DailyPoint, catalog []Event, start time.Time, hori baselineSessions, baselineAmount, predictedSessions, predictedAmount, eventIDs := predictCenter(history, catalog, day) points = append(points, ForecastPoint{ Date: day, BaselineSessions: baselineSessions, BaselineAmountRial: baselineAmount, - PaidSessions: rangeFromResiduals(predictedSessions, sessionResiduals), - PaidAmountRial: rangeFromResiduals(predictedAmount, amountResiduals), EventIDs: eventIDs, + PaidSessions: forecastRange(predictedSessions, sessionResiduals, confidence), + PaidAmountRial: forecastRange(predictedAmount, amountResiduals, confidence), EventIDs: eventIDs, }) } return points @@ -64,6 +61,11 @@ func weekdayMedian(history []DailyPoint, weekday time.Weekday, value func(DailyP values = append(values, value(history[index])) } } + if len(values) == 0 { + for index := len(history) - 1; index >= 0 && len(values) < 8; index-- { + values = append(values, value(history[index])) + } + } return median(values) } @@ -100,6 +102,13 @@ func rangeFromResiduals(center int64, residualValues []int64) Range { } } +func forecastRange(center int64, residualValues []int64, confidence Confidence) Range { + if confidence == ConfidenceLow && len(residualValues) == 0 { + return Range{P10: center / 2, P50: center, P90: center + center/2} + } + return rangeFromResiduals(center, residualValues) +} + func quantile(values []int64, percentile int) int64 { if len(values) == 0 { return 0 @@ -186,7 +195,7 @@ func predictCenter(history []DailyPoint, catalog []Event, day time.Time) (int64, func classifyConfidence(backtest BacktestResult) Confidence { if backtest.Origins < 4 { - return ConfidenceInsufficientData + return ConfidenceLow } worstMAPE := maxInt64(backtest.SessionMAPEBPS, backtest.AmountMAPEBPS) if worstMAPE <= 2_000 && backtest.IntervalCoverageBPS >= 7_000 { @@ -195,7 +204,7 @@ func classifyConfidence(backtest BacktestResult) Confidence { if worstMAPE <= 4_000 && backtest.IntervalCoverageBPS >= 5_000 { return ConfidenceMedium } - return ConfidenceInsufficientData + return ConfidenceLow } type familyUplift struct{ sessions, amount int64 } diff --git a/internal/growth/types.go b/internal/growth/types.go index 761b66f..f21a360 100644 --- a/internal/growth/types.go +++ b/internal/growth/types.go @@ -4,7 +4,7 @@ import "time" const ( CatalogVersion = "1405.2" - AnalyticsVersion = "1.1.0" + AnalyticsVersion = "1.2.0" DefaultModelConfigVersion = "avalai-responses-v1-tools4-turns2-output1600" MinBaselineObservations = 4 @@ -18,6 +18,7 @@ type Confidence string const ( ConfidenceHigh Confidence = "HIGH" ConfidenceMedium Confidence = "MEDIUM" + ConfidenceLow Confidence = "LOW" ConfidenceInsufficientData Confidence = "INSUFFICIENT_DATA" ) diff --git a/internal/growthreport/orchestrator.go b/internal/growthreport/orchestrator.go index 076ea00..76cb02d 100644 --- a/internal/growthreport/orchestrator.go +++ b/internal/growthreport/orchestrator.go @@ -55,7 +55,7 @@ func (s *Service) Create(ctx context.Context, userID, merchantID int64, request }) if err != nil { if errors.Is(err, ErrRefreshLimited) { - next, quotaErr := s.repository.NextRefreshAt(ctx, merchantID, now) + next, quotaErr := s.repository.NextRefreshAt(ctx, merchantID, request.TerminalID, now) if quotaErr != nil { return Record{}, false, quotaErr } @@ -64,18 +64,22 @@ func (s *Service) Create(ctx context.Context, userID, merchantID int64, request return Record{}, false, err } record.Cached = cached - record.NextRefreshAt, err = s.repository.NextRefreshAt(ctx, merchantID, now) + record.NextRefreshAt, err = s.repository.NextRefreshAt(ctx, merchantID, request.TerminalID, 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()) + terminalID := "" + if record.Scope.TerminalID != nil { + terminalID = *record.Scope.TerminalID + } + record.NextRefreshAt, _ = s.repository.NextRefreshAt(ctx, merchantID, terminalID, 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) + record, err := s.repository.Latest(ctx, merchantID, terminalID, growth.AnalyticsVersion) if err == nil { record.Cached = true watermark, start, metadataErr := s.repository.DatasetMetadata(ctx) @@ -84,7 +88,7 @@ func (s *Service) Latest(ctx context.Context, merchantID int64, terminalID strin } else { record.Stale = record.Fingerprint != s.fingerprint(merchantID, terminalID, watermark, start) } - record.NextRefreshAt, _ = s.repository.NextRefreshAt(ctx, merchantID, s.now()) + record.NextRefreshAt, _ = s.repository.NextRefreshAt(ctx, merchantID, terminalID, s.now()) } return record, err } diff --git a/internal/growthreport/repository.go b/internal/growthreport/repository.go index 0635bc7..79bb84a 100644 --- a/internal/growthreport/repository.go +++ b/internal/growthreport/repository.go @@ -90,11 +90,21 @@ func (r *Repository) CreateOrReuse(ctx context.Context, in CreateRecord, beforeI } } 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) + refreshLockKey := fmt.Sprintf("growth-refresh:%d:%s", in.MerchantID, in.TerminalID) + if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,0))`, refreshLockKey); err != nil { + return Record{}, false, fmt.Errorf("locking growth report refresh: %w", err) } - if exists { + var hasAISummary bool + if err := tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM growth_reports WHERE merchant_id=$1 AND terminal_source_key IS NOT DISTINCT FROM $2::text AND status='READY' AND ai_response IS NOT NULL)`, in.MerchantID, nilIfEmpty(in.TerminalID)).Scan(&hasAISummary); err != nil { + return Record{}, false, fmt.Errorf("checking prior AI summary: %w", err) + } + var refreshUsed bool + if hasAISummary { + if err := tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM growth_reports WHERE merchant_id=$1 AND terminal_source_key IS NOT DISTINCT FROM $2::text AND manual_refresh=true AND tehran_refresh_day=$3)`, in.MerchantID, nilIfEmpty(in.TerminalID), in.RefreshDay).Scan(&refreshUsed); err != nil { + return Record{}, false, fmt.Errorf("checking daily refresh: %w", err) + } + } + if refreshUsed { return Record{}, false, ErrRefreshLimited } } @@ -110,9 +120,6 @@ func (r *Repository) CreateOrReuse(ctx context.Context, in CreateRecord, beforeI _, err = tx.Exec(ctx, `INSERT INTO growth_reports(id,merchant_id,terminal_id,terminal_source_key,fingerprint,revision,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,$6,'QUEUED',$7,$8,$9,$10,$11,$12,$13,$14,$15)`, in.ID, in.MerchantID, terminalDBID, nilIfEmpty(in.TerminalID), in.Fingerprint, revision, 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 - } return Record{}, false, fmt.Errorf("creating growth report: %w", err) } if in.IdempotencyKey != "" { @@ -134,8 +141,8 @@ func (r *Repository) Get(ctx context.Context, merchantID int64, id string) (Reco 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) Latest(ctx context.Context, merchantID int64, terminalID, analyticsVersion 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 analytics_version=$3 AND status IN ('READY','PARTIAL') ORDER BY created_at DESC LIMIT 1`, merchantID, nilIfEmpty(terminalID), analyticsVersion)) } func (r *Repository) Claim(ctx context.Context) (Record, error) { @@ -256,18 +263,16 @@ func nilIfEmpty(value string) any { return value } -func isUniqueViolation(err error) bool { - type sqlState interface{ SQLState() string } - var state sqlState - return errors.As(err, &state) && state.SQLState() == "23505" -} - -func (r *Repository) NextRefreshAt(ctx context.Context, merchantID int64, now time.Time) (time.Time, error) { +func (r *Repository) NextRefreshAt(ctx context.Context, merchantID int64, terminalID string, now time.Time) (time.Time, error) { + var hasAISummary bool 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 { + if err := r.pool.QueryRow(ctx, `SELECT + EXISTS(SELECT 1 FROM growth_reports WHERE merchant_id=$1 AND terminal_source_key IS NOT DISTINCT FROM $2::text AND status='READY' AND ai_response IS NOT NULL), + max(created_at) FILTER (WHERE manual_refresh=true) + FROM growth_reports WHERE merchant_id=$1 AND terminal_source_key IS NOT DISTINCT FROM $2::text`, merchantID, nilIfEmpty(terminalID)).Scan(&hasAISummary, &last); err != nil { return time.Time{}, fmt.Errorf("loading refresh quota: %w", err) } - if last == nil { + if !hasAISummary || last == nil { return now.UTC(), nil } return NextTehranRefresh(*last), nil diff --git a/internal/growthreport/repository_integration_test.go b/internal/growthreport/repository_integration_test.go index ba8fd1c..591356b 100644 --- a/internal/growthreport/repository_integration_test.go +++ b/internal/growthreport/repository_integration_test.go @@ -56,6 +56,85 @@ func TestRepositoryChecksUncachedAllowanceBeforeInsert(t *testing.T) { } } +func TestRepositoryAllowsManualRefreshUntilAISummaryExists(t *testing.T) { + pool, merchantID := growthRepositoryTestPool(t) + repository := NewRepository(pool) + ctx := context.Background() + start := time.Now().UTC() + refreshDay := tehranMidnight(start) + newInput := func(id, fingerprint string) CreateRecord { + return CreateRecord{ID: id, MerchantID: merchantID, Fingerprint: fingerprint, DatasetWatermark: "test", ForecastStartDate: &start, CatalogVersion: "test", AnalyticsVersion: "test", PromptVersion: "test", SchemaVersion: "test", Model: "test", ManualRefresh: true, RefreshDay: &refreshDay} + } + + first, _, err := repository.CreateOrReuse(ctx, newInput("growth_aaaaaaaaaaaaaaaaaaaaaaaa", "first"), nil) + if err != nil { + t.Fatal(err) + } + if next, err := repository.NextRefreshAt(ctx, merchantID, "", start); err != nil || next.After(start) { + t.Fatalf("refresh without AI summary should remain available: next=%s error=%v", next, err) + } + second, _, err := repository.CreateOrReuse(ctx, newInput("growth_bbbbbbbbbbbbbbbbbbbbbbbb", "second"), nil) + if err != nil { + t.Fatalf("second refresh without AI summary should be allowed: %v", err) + } + if _, err := pool.Exec(ctx, `UPDATE growth_reports SET status='READY',ai_response=$2 WHERE id=$1`, second.ReportID, `{"briefs":[{"headline":"خلاصه هوشمند"}]}`); err != nil { + t.Fatal(err) + } + _, _, err = repository.CreateOrReuse(ctx, newInput("growth_cccccccccccccccccccccccc", "third"), nil) + if !errors.Is(err, ErrRefreshLimited) { + t.Fatalf("refresh after AI summary error=%v, want ErrRefreshLimited (first=%s)", err, first.ReportID) + } + if next, err := repository.NextRefreshAt(ctx, merchantID, "", start); err != nil || !next.After(start) { + t.Fatalf("refresh after AI summary should have a future boundary: next=%s error=%v", next, err) + } +} + +func TestRepositoryLatestIgnoresReportsFromOlderAnalyticsVersions(t *testing.T) { + pool, merchantID := growthRepositoryTestPool(t) + repository := NewRepository(pool) + ctx := context.Background() + now := time.Now().UTC() + + _, err := pool.Exec(ctx, `INSERT INTO growth_reports(id,merchant_id,fingerprint,status,forecast_start_date,dataset_watermark,catalog_version,analytics_version,prompt_version,schema_version,model,deterministic_evidence,created_at,updated_at) VALUES($1,$2,'old-fingerprint','PARTIAL',current_date,'test','test','1.1.0','test','test','test',$3,$4,$4)`, "growth_dddddddddddddddddddddddd", merchantID, `{"forecast":{"confidence":"INSUFFICIENT_DATA","points":[]}}`, now) + if err != nil { + t.Fatal(err) + } + + if _, err := repository.Latest(ctx, merchantID, "", "1.2.0"); !errors.Is(err, ErrNotFound) { + t.Fatalf("Latest error=%v, want ErrNotFound for a stale analytics version", err) + } +} + +func TestRepositoryRefreshQuotaIsScopedPerTerminal(t *testing.T) { + pool, merchantID := growthRepositoryTestPool(t) + repository := NewRepository(pool) + ctx := context.Background() + start := time.Now().UTC() + refreshDay := tehranMidnight(start) + for _, terminal := range []string{"terminal-a", "terminal-b"} { + if _, err := pool.Exec(ctx, `INSERT INTO terminals(merchant_id,source_key,verify_type) VALUES($1,$2,'test')`, merchantID, terminal); err != nil { + t.Fatal(err) + } + } + newInput := func(id, terminal string) CreateRecord { + return CreateRecord{ID: id, MerchantID: merchantID, TerminalID: terminal, Fingerprint: id, DatasetWatermark: "test", ForecastStartDate: &start, CatalogVersion: "test", AnalyticsVersion: "test", PromptVersion: "test", SchemaVersion: "test", Model: "test", ManualRefresh: true, RefreshDay: &refreshDay} + } + + first, _, err := repository.CreateOrReuse(ctx, newInput("growth_eeeeeeeeeeeeeeeeeeeeeeee", "terminal-a"), nil) + if err != nil { + t.Fatal(err) + } + if _, err := pool.Exec(ctx, `UPDATE growth_reports SET status='READY',ai_response=$2 WHERE id=$1`, first.ReportID, `{"briefs":[{"headline":"خلاصه هوشمند"}]}`); err != nil { + t.Fatal(err) + } + if _, _, err := repository.CreateOrReuse(ctx, newInput("growth_ffffffffffffffffffffffff", "terminal-a"), nil); !errors.Is(err, ErrRefreshLimited) { + t.Fatalf("same-terminal refresh error=%v, want ErrRefreshLimited", err) + } + if _, _, err := repository.CreateOrReuse(ctx, newInput("growth_111111111111111111111111", "terminal-b"), nil); err != nil { + t.Fatalf("another terminal must retain its own refresh quota: %v", err) + } +} + func growthRepositoryTestPool(t *testing.T) (*pgxpool.Pool, int64) { t.Helper() databaseURL := os.Getenv("TEST_DATABASE_URL") @@ -74,6 +153,7 @@ func growthRepositoryTestPool(t *testing.T) (*pgxpool.Pool, int64) { } t.Cleanup(func() { _, _ = pool.Exec(context.Background(), `DELETE FROM growth_reports WHERE merchant_id=$1`, merchantID) + _, _ = pool.Exec(context.Background(), `DELETE FROM terminals WHERE merchant_id=$1`, merchantID) _, _ = pool.Exec(context.Background(), `DELETE FROM merchants WHERE id=$1`, merchantID) pool.Close() }) diff --git a/migrations/000003_ai_summary_refresh.sql b/migrations/000003_ai_summary_refresh.sql new file mode 100644 index 0000000..9c3ea6b --- /dev/null +++ b/migrations/000003_ai_summary_refresh.sql @@ -0,0 +1,13 @@ +DROP INDEX IF EXISTS growth_reports_daily_merchant_refresh_idx; + +-- statement-break +CREATE INDEX IF NOT EXISTS growth_reports_scope_refresh_day_idx + ON growth_reports (merchant_id, terminal_source_key, tehran_refresh_day, created_at DESC) + WHERE manual_refresh = true; + +-- The application serializes scope-specific refresh decisions with a PostgreSQL +-- advisory transaction lock. A refresh consumes the daily slot only after the +-- requested scope has a persisted READY report with AI copy. + +-- statement-break +INSERT INTO schema_migrations(version) VALUES (3) ON CONFLICT DO NOTHING;