feat: orchestrate deterministic growth report generation
This commit is contained in:
@@ -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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user