From dd0ae15a95b55a8a75337dbc5207380393886edc Mon Sep 17 00:00:00 2001 From: Amirhossein Khalili Date: Fri, 21 Aug 2026 00:43:31 +0330 Subject: [PATCH] fix: preserve typed growth report revisions --- internal/growthreport/model.go | 11 ++++++++++- internal/growthreport/repository.go | 25 +++++++++++++++++-------- migrations/000002_growth_analyst.sql | 4 ++-- 3 files changed, 29 insertions(+), 11 deletions(-) diff --git a/internal/growthreport/model.go b/internal/growthreport/model.go index 4f8c48d..16a36aa 100644 --- a/internal/growthreport/model.go +++ b/internal/growthreport/model.go @@ -2,6 +2,7 @@ package growthreport import ( "encoding/json" + "fmt" "time" ) @@ -44,11 +45,12 @@ type Usage struct { type Report struct { ReportID string `json:"reportId"` + Revision int `json:"revision"` Status Status `json:"status"` Scope Scope `json:"scope"` HorizonDays int `json:"horizonDays"` ForecastStartDate *time.Time `json:"forecastStartDate,omitempty"` - DeterministicEvidence json.RawMessage `json:"deterministicEvidence"` + DeterministicEvidence json.RawMessage `json:"deterministicEvidence,omitempty"` AICopy *AICopy `json:"aiCopy,omitempty"` FailureCode *string `json:"failureCode,omitempty"` Cached bool `json:"cached"` @@ -79,3 +81,10 @@ type CreateRequest struct { TerminalID string `json:"terminalId"` Refresh bool `json:"refresh"` } + +type RefreshLimitError struct{ NextRefreshAt time.Time } + +func (e *RefreshLimitError) Error() string { + return fmt.Sprintf("%s: next refresh at %s", ErrRefreshLimited, e.NextRefreshAt.Format(time.RFC3339)) +} +func (e *RefreshLimitError) Unwrap() error { return ErrRefreshLimited } diff --git a/internal/growthreport/repository.go b/internal/growthreport/repository.go index 87f0e12..7136cbd 100644 --- a/internal/growthreport/repository.go +++ b/internal/growthreport/repository.go @@ -46,10 +46,7 @@ func (r *Repository) CreateOrReuse(ctx context.Context, in CreateRecord) (Record 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)) - } + lockKey := fmt.Sprintf("growth-revision:%d:%s", in.MerchantID, in.TerminalID) 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) } @@ -99,8 +96,12 @@ func (r *Repository) CreateOrReuse(ctx context.Context, in CreateRecord) (Record 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) + var revision int + if err := tx.QueryRow(ctx, `SELECT coalesce(max(revision),0)+1 FROM growth_reports WHERE merchant_id=$1 AND terminal_source_key IS NOT DISTINCT FROM $2::text`, in.MerchantID, nilIfEmpty(in.TerminalID)).Scan(&revision); err != nil { + return Record{}, false, fmt.Errorf("calculating growth report revision: %w", err) + } + _, 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 @@ -181,7 +182,7 @@ func (r *Repository) Complete(ctx context.Context, reportID string, status Statu 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` +const reportSelect = `SELECT id,revision,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 { @@ -201,7 +202,7 @@ func scanRecord(row rowScanner) (Record, error) { 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) + err := row.Scan(&r.ReportID, &r.Revision, &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 } @@ -258,3 +259,11 @@ func (r *Repository) DatasetMetadata(ctx context.Context) (string, time.Time, er } return sha + ":" + latest.Format(time.DateOnly), latest.AddDate(0, 0, 1), nil } + +func (r *Repository) CleanupSuperseded(ctx context.Context, cutoff time.Time) (int64, error) { + command, err := r.pool.Exec(ctx, `DELETE FROM growth_reports old WHERE old.created_at<$1 AND EXISTS(SELECT 1 FROM growth_reports newer WHERE newer.merchant_id=old.merchant_id AND newer.terminal_source_key IS NOT DISTINCT FROM old.terminal_source_key AND newer.id<>old.id AND newer.revision>old.revision AND newer.status IN ('READY','PARTIAL'))`, cutoff) + if err != nil { + return 0, fmt.Errorf("cleaning superseded growth reports: %w", err) + } + return command.RowsAffected(), nil +} diff --git a/migrations/000002_growth_analyst.sql b/migrations/000002_growth_analyst.sql index f12b35f..7bc8335 100644 --- a/migrations/000002_growth_analyst.sql +++ b/migrations/000002_growth_analyst.sql @@ -14,7 +14,7 @@ CREATE TABLE IF NOT EXISTS growth_reports ( prompt_version text NOT NULL, schema_version text NOT NULL, model text NOT NULL, - deterministic_evidence jsonb NOT NULL DEFAULT '{}'::jsonb, + deterministic_evidence jsonb, ai_response jsonb, failure_code text, input_tokens bigint NOT NULL DEFAULT 0 CHECK (input_tokens >= 0), @@ -36,7 +36,7 @@ CREATE UNIQUE INDEX IF NOT EXISTS growth_reports_daily_merchant_refresh_idx WHERE manual_refresh = true; -- statement-break CREATE INDEX IF NOT EXISTS growth_reports_scope_latest_idx - ON growth_reports (merchant_id, terminal_id, created_at DESC); + ON growth_reports (merchant_id, terminal_id, revision DESC, created_at DESC); -- statement-break CREATE INDEX IF NOT EXISTS growth_reports_pending_idx ON growth_reports (created_at)