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

fix: keep growth analysis available per terminal

Browse Source
This commit is contained in:
Amirhossein Khalili
2026-08-21 02:57:46 +03:30
parent 7951664d08
commit e6afaa7a10
11 changed files with 180 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
internal/growthreport/orchestrator.go
View File

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

41
internal/growthreport/repository.go
View File

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

80
internal/growthreport/repository_integration_test.go
View File

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