146 lines
8.7 KiB
Go
146 lines
8.7 KiB
Go
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"
|
|
}
|
|
|