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

fix: preserve typed growth report revisions

Browse Source
This commit is contained in:
Amirhossein Khalili
2026-08-21 00:43:31 +03:30
parent 49084ea577
commit dd0ae15a95
3 changed files with 29 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
internal/growthreport/model.go
View File

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

25
internal/growthreport/repository.go
View File

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

4
migrations/000002_growth_analyst.sql
View File

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