package analytics import ( "context" "fmt" "math" "slices" "sync" "time" "github.com/jackc/pgx/v5/pgxpool" ) type Period struct { PeriodID string `json:"periodId"` PeriodType string `json:"periodType"` StartAt time.Time `json:"startAt"` EndAt time.Time `json:"endAt"` LocalStartAt time.Time `json:"localStartAt"` LocalEndAt time.Time `json:"localEndAt"` Timezone string `json:"timezone"` JalaliLabelFA string `json:"jalaliLabelFa"` IsComplete bool `json:"isComplete"` IsDefault bool `json:"isDefault"` } type Run struct { RunID string `json:"runId"` State string `json:"state"` Period Period `json:"period"` DatasetSnapshotID string `json:"datasetSnapshotId"` MetricDefinitionVersion string `json:"metricDefinitionVersion"` CreatedAt time.Time `json:"createdAt"` UpdatedAt time.Time `json:"updatedAt"` EventsURL string `json:"eventsUrl"` Scope RunScope `json:"scope"` Sections map[string]any `json:"sections"` } type RunScope struct { TerminalID *string `json:"terminalId"` } type PaymentHistoryPoint struct { LocalDate string `json:"localDate"` TotalSessions int64 `json:"totalSessions"` BankPaidSessions int64 `json:"bankPaidSessions"` BankPaidAmountRial int64 `json:"bankPaidAmountRial"` } type Service struct { pool *pgxpool.Pool mu sync.RWMutex runs map[string]storedRun } type storedRun struct { merchantID int64 run Run } func New(pool *pgxpool.Pool) *Service { return &Service{pool: pool, runs: map[string]storedRun{}} } func (s *Service) Periods(ctx context.Context) ([]Period, error) { var start, end time.Time err := s.pool.QueryRow(ctx, `SELECT min(created_at),max(created_at) FROM payment_sessions`).Scan(&start, &end) if err != nil { return nil, fmt.Errorf("loading dataset period: %w", err) } location, err := time.LoadLocation("Asia/Tehran") if err != nil { return nil, fmt.Errorf("loading tehran timezone: %w", err) } return latestCompletePeriods(start, end, location), nil } func (s *Service) Start(ctx context.Context, merchantID int64, period Period, terminalID string) (Run, error) { if err := s.validateTerminal(ctx, merchantID, terminalID); err != nil { return Run{}, err } key := fmt.Sprintf("%d:%s:%s", merchantID, period.PeriodID, terminalID) s.mu.RLock() existing, ok := s.runs[key] s.mu.RUnlock() if ok { return existing.run, nil } now := time.Now().UTC() run := Run{RunID: fmt.Sprintf("run_%d%d", merchantID, now.UnixNano()), State: "COMPLETE", Period: period, DatasetSnapshotID: "ds_localimport", MetricDefinitionVersion: "0.4.0", CreatedAt: now, UpdatedAt: now, Sections: map[string]any{}} if terminalID != "" { run.Scope.TerminalID = &terminalID } run.EventsURL = "/api/v1/analytics/runs/" + run.RunID + "/events" metrics, err := s.metrics(ctx, merchantID, period, terminalID) if err != nil { return Run{}, err } history, err := s.paymentHistory(ctx, merchantID, period, terminalID) if err != nil { return Run{}, err } metadata := map[string]any{"calculatedAt": now, "datasetSnapshotId": run.DatasetSnapshotID, "metricDefinitionVersion": "0.4.0"} run.Sections["overview"] = map[string]any{"sectionId": "overview", "state": "READY", "metadata": metadata, "data": metrics.overview(period, history)} run.Sections["paymentPerformance"] = map[string]any{"sectionId": "payment_performance", "state": "READY", "metadata": metadata, "data": metrics.performance()} run.Sections["fees"] = map[string]any{"sectionId": "fees", "state": "READY", "metadata": metadata, "data": metrics.fees(period)} peerCount, peerSessions, err := s.peerEligibility(ctx, merchantID, period) if err != nil { return Run{}, err } if peerCount < 10 || peerSessions < 1000 { run.Sections["peerComparison"] = map[string]any{"sectionId": "peer_comparison", "state": "SUPPRESSED", "metadata": metadata, "reason": "INSUFFICIENT_COHORT", "eligiblePeerCount": peerCount, "combinedPeerSessionCount": peerSessions, "requiredPeerCount": 10, "requiredPeerSessionCount": 1000} } else { peerMetrics, err := s.peerMetrics(ctx, merchantID, period) if err != nil { return Run{}, err } run.Sections["peerComparison"] = map[string]any{"sectionId": "peer_comparison", "state": "READY", "metadata": metadata, "data": map[string]any{"eligiblePeerCount": peerCount, "combinedPeerSessionCount": peerSessions, "metrics": peerMetrics}} } s.mu.Lock() s.runs[key] = storedRun{merchantID: merchantID, run: run} s.runs[run.RunID] = storedRun{merchantID: merchantID, run: run} s.mu.Unlock() return run, nil } func (s *Service) Run(merchantID int64, runID string) (Run, bool) { s.mu.RLock() stored, ok := s.runs[runID] s.mu.RUnlock() return stored.run, ok && stored.merchantID == merchantID } type metricValues struct { total, paid, verified, unverified, failed, reversed, noAttempt, reachedBank, retried, recovered int64 paidAmount, feeAmount int64 initP50, initP95, verifyP50, verifyP95 *float64 initSamples, verifySamples int64 } func (s *Service) metrics(ctx context.Context, merchantID int64, p Period, terminalID string) (metricValues, error) { var m metricValues query := `WITH scoped AS ( SELECT ps.*,count(pa.id) attempts,bool_or(pa.status='InBank') reached_bank FROM payment_sessions ps LEFT JOIN payment_attempts pa ON pa.session_id=ps.id WHERE ps.merchant_id=$1 AND ps.created_at >= $2 AND ps.created_at < $3 AND ($4='' OR ps.terminal_id=(SELECT id FROM terminals WHERE merchant_id=$1 AND source_key=$4)) GROUP BY ps.id ), timings AS ( SELECT percentile_disc(0.5) within group(order by pa.init_time_ms)::float8 init_p50, percentile_disc(0.95) within group(order by pa.init_time_ms)::float8 init_p95, percentile_disc(0.5) within group(order by pa.verify_time_ms)::float8 verify_p50, percentile_disc(0.95) within group(order by pa.verify_time_ms)::float8 verify_p95, count(pa.init_time_ms) init_samples,count(pa.verify_time_ms) verify_samples FROM payment_attempts pa JOIN payment_sessions ps ON ps.id=pa.session_id WHERE ps.merchant_id=$1 AND ps.created_at >= $2 AND ps.created_at < $3 AND ($4='' OR ps.terminal_id=(SELECT id FROM terminals WHERE merchant_id=$1 AND source_key=$4))) SELECT count(*),count(*) filter(where status in ('Paid','Verified')), count(*) filter(where status='Verified'),count(*) filter(where status='Paid'),count(*) filter(where status='Failed'), count(*) filter(where status='Reversed'),count(*) filter(where attempts=0),count(*) filter(where reached_bank), count(*) filter(where attempts>1),count(*) filter(where attempts>1 and status in ('Paid','Verified')), coalesce(sum(amount_rial) filter(where status in ('Paid','Verified')),0),coalesce(sum(fee_amount_rial) filter(where status in ('Paid','Verified')),0), t.init_p50,t.init_p95,t.verify_p50,t.verify_p95,t.init_samples,t.verify_samples FROM scoped CROSS JOIN timings t GROUP BY t.init_p50,t.init_p95,t.verify_p50,t.verify_p95,t.init_samples,t.verify_samples` err := s.pool.QueryRow(ctx, query, merchantID, p.StartAt, p.EndAt, terminalID).Scan(&m.total, &m.paid, &m.verified, &m.unverified, &m.failed, &m.reversed, &m.noAttempt, &m.reachedBank, &m.retried, &m.recovered, &m.paidAmount, &m.feeAmount, &m.initP50, &m.initP95, &m.verifyP50, &m.verifyP95, &m.initSamples, &m.verifySamples) if err != nil { return metricValues{}, fmt.Errorf("calculating merchant metrics: %w", err) } return m, nil } func integerMetric(id string, value int64, unit string) map[string]any { return map[string]any{"metricId": id, "value": value, "unit": unit} } func decimalMetric(id string, numerator, denominator int64, unit string) map[string]any { var value any if denominator > 0 { value = fmt.Sprintf("%.2f", float64(numerator)*100/float64(denominator)) } return map[string]any{"metricId": id, "value": value, "unit": unit, "numerator": numerator, "denominator": denominator} } func unavailableComparison(period Period) map[string]any { return map[string]any{"state": "UNAVAILABLE", "baselinePeriod": period, "absoluteChange": nil, "relativeChangePercent": nil, "unavailableReason": "SOURCE_UNAVAILABLE", "isActionableInsightEligible": false} } func (m metricValues) overview(period Period, history []PaymentHistoryPoint) map[string]any { return map[string]any{ "bankPaidSessions": integerMetric("bank_paid_sessions", m.paid, "SESSIONS"), "bankPaidAmountRial": integerMetric("bank_paid_amount_rial", m.paidAmount, "RIAL"), "paymentSuccessRate": decimalMetric("payment_success_rate", m.paid, m.total, "PERCENT"), "verificationCompletionRate": decimalMetric("verification_completion_rate", m.verified, m.paid, "PERCENT"), "comparisons": map[string]any{"bank_paid_sessions": unavailableComparison(period), "bank_paid_amount_rial": unavailableComparison(period), "payment_success_rate": unavailableComparison(period), "verification_completion_rate": unavailableComparison(period)}, "paymentHistory": history, } } func (s *Service) paymentHistory(ctx context.Context, merchantID int64, period Period, terminalID string) ([]PaymentHistoryPoint, error) { rows, err := s.pool.Query(ctx, `SELECT (ps.created_at AT TIME ZONE 'Asia/Tehran')::date local_date, count(*),count(*) FILTER (WHERE ps.status IN ('Paid','Verified')), coalesce(sum(ps.amount_rial) FILTER (WHERE ps.status IN ('Paid','Verified')),0) FROM payment_sessions ps WHERE ps.merchant_id=$1 AND ps.created_at >= $2 AND ps.created_at < $3 AND ($4='' OR ps.terminal_id=(SELECT id FROM terminals WHERE merchant_id=$1 AND source_key=$4)) GROUP BY local_date ORDER BY local_date`, merchantID, period.StartAt, period.EndAt, terminalID) if err != nil { return nil, fmt.Errorf("querying payment history: %w", err) } defer rows.Close() points := make([]PaymentHistoryPoint, 0, 93) for rows.Next() { var date time.Time var point PaymentHistoryPoint if err := rows.Scan(&date, &point.TotalSessions, &point.BankPaidSessions, &point.BankPaidAmountRial); err != nil { return nil, fmt.Errorf("scanning payment history: %w", err) } point.LocalDate = date.Format(time.DateOnly) points = append(points, point) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("iterating payment history: %w", err) } return points, nil } func timing(id string, p50, p95 *float64, samples, eligible int64) map[string]any { var a, b any if p50 != nil { a = int64(math.Round(*p50)) } if p95 != nil { b = int64(math.Round(*p95)) } var coverage any if eligible > 0 { coverage = fmt.Sprintf("%.2f", float64(samples)*100/float64(eligible)) } return map[string]any{"metricId": id, "p50Ms": a, "p95Ms": b, "sampleSize": samples, "eligibleSampleSize": eligible, "coveragePercent": coverage} } func (m metricValues) performance() map[string]any { return map[string]any{ "bankPaidSessions": integerMetric("bank_paid_sessions", m.paid, "SESSIONS"), "verifiedSessions": integerMetric("verified_sessions", m.verified, "SESSIONS"), "unverifiedPaidSessions": integerMetric("unverified_paid_sessions", m.unverified, "SESSIONS"), "failedSessions": integerMetric("failed_sessions", m.failed, "SESSIONS"), "reversedSessions": integerMetric("reversed_sessions", m.reversed, "SESSIONS"), "noAttemptSessions": integerMetric("no_attempt_sessions", m.noAttempt, "SESSIONS"), "reachedBankSessions": integerMetric("reached_bank_sessions", m.reachedBank, "SESSIONS"), "retriedSessions": integerMetric("retried_sessions", m.retried, "SESSIONS"), "retryRecoveryRate": decimalMetric("retry_recovery_rate", m.recovered, m.retried, "PERCENT"), "initTiming": timing("init_time_ms", m.initP50, m.initP95, m.initSamples, m.total), "verifyTiming": timing("verify_time_ms", m.verifyP50, m.verifyP95, m.verifySamples, m.paid), } } func (m metricValues) fees(period Period) map[string]any { return map[string]any{"paidFeeAmountRial": integerMetric("paid_fee_amount_rial", m.feeAmount, "RIAL"), "bankPaidAmountRial": integerMetric("bank_paid_amount_rial", m.paidAmount, "RIAL"), "paidFeeRate": decimalMetric("paid_fee_rate", m.feeAmount, m.paidAmount, "RATIO"), "comparison": unavailableComparison(period)} } func (s *Service) peerEligibility(ctx context.Context, merchantID int64, p Period) (int64, int64, error) { var peers, sessions int64 err := s.pool.QueryRow(ctx, `SELECT count(distinct ps.merchant_id),count(*) FROM payment_sessions ps JOIN merchants m ON m.id=ps.merchant_id WHERE m.category_id=(SELECT category_id FROM merchants WHERE id=$1) AND ps.merchant_id<>$1 AND ps.created_at>=$2 AND ps.created_at<$3`, merchantID, p.StartAt, p.EndAt).Scan(&peers, &sessions) if err != nil { return 0, 0, fmt.Errorf("calculating peer eligibility: %w", err) } return peers, sessions, nil } type peerRow struct { merchantID int64 values [8]*float64 } func (s *Service) peerMetrics(ctx context.Context, merchantID int64, p Period) ([]map[string]any, error) { query := `WITH session_stats AS (SELECT ps.merchant_id,ps.id,ps.status,ps.amount_rial,ps.fee_amount_rial,count(pa.id) attempts FROM payment_sessions ps LEFT JOIN payment_attempts pa ON pa.session_id=ps.id JOIN merchants m ON m.id=ps.merchant_id WHERE m.category_id=(SELECT category_id FROM merchants WHERE id=$1) AND ps.created_at>=$2 AND ps.created_at<$3 GROUP BY ps.id), aggregate AS (SELECT merchant_id,100.0*count(*) filter(where status in ('Paid','Verified'))/nullif(count(*),0) success, 100.0*count(*) filter(where status='Verified')/nullif(count(*) filter(where status in ('Paid','Verified')),0) verification, 100.0*count(*) filter(where attempts>1 and status in ('Paid','Verified'))/nullif(count(*) filter(where attempts>1),0) recovery, 100.0*sum(fee_amount_rial) filter(where status in ('Paid','Verified'))/nullif(sum(amount_rial) filter(where status in ('Paid','Verified')),0) fee FROM session_stats GROUP BY merchant_id), timing AS (SELECT ps.merchant_id,percentile_disc(.5) within group(order by pa.init_time_ms)::float8 ip50,percentile_disc(.95) within group(order by pa.init_time_ms)::float8 ip95, percentile_disc(.5) within group(order by pa.verify_time_ms)::float8 vp50,percentile_disc(.95) within group(order by pa.verify_time_ms)::float8 vp95 FROM payment_sessions ps JOIN payment_attempts pa ON pa.session_id=ps.id JOIN merchants m ON m.id=ps.merchant_id WHERE m.category_id=(SELECT category_id FROM merchants WHERE id=$1) AND ps.created_at>=$2 AND ps.created_at<$3 GROUP BY ps.merchant_id) SELECT a.merchant_id,a.success::float8,a.verification::float8,a.recovery::float8,t.ip50,t.ip95,t.vp50,t.vp95,a.fee::float8 FROM aggregate a JOIN timing t USING(merchant_id)` rows, err := s.pool.Query(ctx, query, merchantID, p.StartAt, p.EndAt) if err != nil { return nil, fmt.Errorf("calculating peer metrics: %w", err) } defer rows.Close() all := []peerRow{} for rows.Next() { var row peerRow if err := rows.Scan(&row.merchantID, &row.values[0], &row.values[1], &row.values[2], &row.values[3], &row.values[4], &row.values[5], &row.values[6], &row.values[7]); err != nil { return nil, fmt.Errorf("scanning peer metrics: %w", err) } all = append(all, row) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("iterating peer metrics: %w", err) } ids := []string{"payment_success_rate", "verification_completion_rate", "retry_recovery_rate", "init_time_ms_p50", "init_time_ms_p95", "verify_time_ms_p50", "verify_time_ms_p95", "paid_fee_rate"} metrics := make([]map[string]any, 0, 8) for index, id := range ids { active := 0.0 peers := []float64{} for _, row := range all { if row.values[index] == nil { continue } if row.merchantID == merchantID { active = *row.values[index] } else { peers = append(peers, *row.values[index]) } } if len(peers) == 0 { peers = append(peers, 0) } slices.Sort(peers) median := peers[len(peers)/2] if len(peers)%2 == 0 { median = (peers[len(peers)/2-1] + median) / 2 } lowerIsBetter := index >= 3 better := 0 for _, value := range peers { if (!lowerIsBetter && active >= value) || (lowerIsBetter && active <= value) { better++ } } percentile := float64(better) * 100 / float64(len(peers)) direction := "HIGHER_IS_BETTER" if lowerIsBetter { direction = "LOWER_IS_BETTER" } metrics = append(metrics, map[string]any{"metricId": id, "cohortMedian": fmt.Sprintf("%.2f", median), "merchantPerformancePercentile": fmt.Sprintf("%.2f", percentile), "direction": direction}) } return metrics, nil }