Amirhossein Khalili
d6de48fd15
Keep JSON and bounded CSV exports on one canonical predicate, replace replayable offsets with encrypted tenant/filter-bound keyset cursors, and suppress merchant-grain peers under terminal scope. Validate period filters and align OpenAPI and capability specs with runtime behavior.
273 lines
11 KiB
Go
273 lines
11 KiB
Go
package analytics
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
"strings"
|
||
"time"
|
||
|
||
"github.com/jackc/pgx/v5"
|
||
)
|
||
|
||
const MaxExportRows = 100_000
|
||
|
||
var ErrExportTooLarge = errors.New("analytics: export exceeds row limit")
|
||
|
||
const detailFilterPredicate = `
|
||
ps.merchant_id=$1 AND ps.created_at>=$2 AND ps.created_at<$3
|
||
AND ($4='ALL'
|
||
OR ($4='UNVERIFIED_PAID' AND ps.status='Paid')
|
||
OR ($4='FAILED' AND ps.status='Failed')
|
||
OR ($4='RETRIED' AND (SELECT count(*) FROM payment_attempts retry_attempts WHERE retry_attempts.session_id=ps.id)>1))
|
||
AND ($5='' OR t.source_key=$5)
|
||
AND ($6='' OR
|
||
($6='NO_ATTEMPT' AND NOT EXISTS (SELECT 1 FROM payment_attempts failure_attempts WHERE failure_attempts.session_id=ps.id)) OR
|
||
($6='REACHED_BANK' AND EXISTS (SELECT 1 FROM payment_attempts failure_attempts WHERE failure_attempts.session_id=ps.id AND failure_attempts.status='InBank')) OR
|
||
($6='OTHER_OBSERVED' AND EXISTS (SELECT 1 FROM payment_attempts failure_attempts WHERE failure_attempts.session_id=ps.id) AND NOT EXISTS (SELECT 1 FROM payment_attempts failure_attempts WHERE failure_attempts.session_id=ps.id AND failure_attempts.status='InBank')))
|
||
AND ($7='' OR ps.status=$7)
|
||
AND ($8::bigint IS NULL OR ps.amount_rial >= $8)
|
||
AND ($9::bigint IS NULL OR ps.amount_rial <= $9)`
|
||
|
||
type DetailFilter struct {
|
||
View string
|
||
FailureStage string
|
||
TerminalID string
|
||
Status string
|
||
MinAmountRial *int64
|
||
MaxAmountRial *int64
|
||
Cursor string
|
||
Limit int
|
||
}
|
||
|
||
func (f DetailFilter) Validate() error {
|
||
views := map[string]bool{"ALL": true, "UNVERIFIED_PAID": true, "FAILED": true, "RETRIED": true}
|
||
if !views[f.View] {
|
||
return errors.New("analytics: invalid detail view")
|
||
}
|
||
statuses := map[string]bool{"": true, "Paid": true, "Verified": true, "Failed": true, "Reversed": true}
|
||
if !statuses[f.Status] {
|
||
return errors.New("analytics: invalid session status")
|
||
}
|
||
stages := map[string]bool{"": true, "NO_ATTEMPT": true, "REACHED_BANK": true, "OTHER_OBSERVED": true}
|
||
if !stages[f.FailureStage] || (f.FailureStage != "" && f.View != "FAILED") {
|
||
return errors.New("analytics: invalid failure stage")
|
||
}
|
||
if f.MinAmountRial != nil && *f.MinAmountRial < 0 || f.MaxAmountRial != nil && *f.MaxAmountRial < 0 {
|
||
return errors.New("analytics: amount bounds must be non-negative")
|
||
}
|
||
if f.MinAmountRial != nil && f.MaxAmountRial != nil && *f.MinAmountRial > *f.MaxAmountRial {
|
||
return errors.New("analytics: minimum amount exceeds maximum")
|
||
}
|
||
if f.Limit < 1 || f.Limit > 100 {
|
||
return errors.New("analytics: detail limit outside range")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
type Attempt struct {
|
||
Sequence int `json:"sequence"`
|
||
Status string `json:"status"`
|
||
CreatedAt *time.Time `json:"createdAt"`
|
||
InitTimeMS *int `json:"initTimeMs"`
|
||
VerifyTimeMS *int `json:"verifyTimeMs"`
|
||
}
|
||
type Detail struct {
|
||
SessionID string `json:"sessionId"`
|
||
TerminalID string `json:"terminalId"`
|
||
SessionStatus string `json:"sessionStatus"`
|
||
AmountRial int64 `json:"amountRial"`
|
||
FeeAmountRial int64 `json:"feeAmountRial"`
|
||
CreatedAt time.Time `json:"createdAt"`
|
||
ExpiresAt *time.Time `json:"expiresAt"`
|
||
VerifiedAt *time.Time `json:"verifiedAt"`
|
||
SettledAt *time.Time `json:"settledAt"`
|
||
AttemptCount int `json:"attemptCount"`
|
||
FailureStage *string `json:"failureStage"`
|
||
Attempts []Attempt `json:"attempts"`
|
||
}
|
||
|
||
type ExportDetail struct {
|
||
Detail
|
||
LastAttemptStatus *string
|
||
LastAttemptAt *time.Time
|
||
}
|
||
|
||
func (s *Service) Details(ctx context.Context, merchantID int64, period Period, filter DetailFilter) ([]Detail, string, error) {
|
||
if filter.Limit <= 0 {
|
||
filter.Limit = 25
|
||
}
|
||
if err := filter.Validate(); err != nil {
|
||
return nil, "", err
|
||
}
|
||
var cursorCreatedAt *time.Time
|
||
var cursorSessionID int64
|
||
if filter.Cursor != "" {
|
||
position, err := s.decodeDetailCursor(merchantID, period, filter, filter.Cursor)
|
||
if err != nil {
|
||
return nil, "", err
|
||
}
|
||
cursorCreatedAt = &position.CreatedAt
|
||
cursorSessionID = position.SessionID
|
||
}
|
||
query := `SELECT ps.id,ps.source_key,t.source_key,ps.status,ps.amount_rial,ps.fee_amount_rial,ps.created_at,ps.expires_at,ps.verified_at,ps.settled_at,
|
||
count(pa.id)::int,coalesce(bool_or(pa.status='InBank'),false) FROM payment_sessions ps JOIN terminals t ON t.id=ps.terminal_id LEFT JOIN payment_attempts pa ON pa.session_id=ps.id
|
||
WHERE ` + detailFilterPredicate + `
|
||
AND ($10::timestamptz IS NULL OR ps.created_at < $10 OR (ps.created_at=$10 AND ps.id<$11))
|
||
GROUP BY ps.id,t.source_key ORDER BY ps.created_at DESC,ps.id DESC LIMIT $12`
|
||
rows, err := s.pool.Query(ctx, query, merchantID, period.StartAt, period.EndAt, filter.View, filter.TerminalID, filter.FailureStage, filter.Status, filter.MinAmountRial, filter.MaxAmountRial, cursorCreatedAt, cursorSessionID, filter.Limit+1)
|
||
if err != nil {
|
||
return nil, "", fmt.Errorf("querying session details: %w", err)
|
||
}
|
||
defer rows.Close()
|
||
details := make([]Detail, 0, filter.Limit)
|
||
ids := make([]int64, 0, filter.Limit)
|
||
hasMore := false
|
||
for rows.Next() {
|
||
var id int64
|
||
var reachedBank bool
|
||
var d Detail
|
||
if err := rows.Scan(&id, &d.SessionID, &d.TerminalID, &d.SessionStatus, &d.AmountRial, &d.FeeAmountRial, &d.CreatedAt, &d.ExpiresAt, &d.VerifiedAt, &d.SettledAt, &d.AttemptCount, &reachedBank); err != nil {
|
||
return nil, "", fmt.Errorf("scanning session detail: %w", err)
|
||
}
|
||
d.SessionID = "ses_" + d.SessionID
|
||
d.Attempts = []Attempt{}
|
||
setFailureStage(&d, reachedBank)
|
||
if len(details) == filter.Limit {
|
||
hasMore = true
|
||
break
|
||
}
|
||
details = append(details, d)
|
||
ids = append(ids, id)
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return nil, "", fmt.Errorf("iterating session details: %w", err)
|
||
}
|
||
for i, id := range ids {
|
||
attempts, err := s.attempts(ctx, id)
|
||
if err != nil {
|
||
return nil, "", err
|
||
}
|
||
details[i].Attempts = attempts
|
||
}
|
||
next := ""
|
||
if hasMore {
|
||
next, err = s.encodeDetailCursor(merchantID, period, filter, details[len(details)-1].CreatedAt, ids[len(ids)-1])
|
||
if err != nil {
|
||
return nil, "", fmt.Errorf("encoding detail cursor: %w", err)
|
||
}
|
||
}
|
||
return details, next, nil
|
||
}
|
||
|
||
func (s *Service) ExportDetails(ctx context.Context, merchantID int64, period Period, filter DetailFilter) ([]ExportDetail, error) {
|
||
filter.Cursor = ""
|
||
filter.Limit = 100
|
||
if err := filter.Validate(); err != nil {
|
||
return nil, err
|
||
}
|
||
query := `SELECT ps.id,ps.source_key,t.source_key,ps.status,ps.amount_rial,ps.fee_amount_rial,ps.created_at,ps.expires_at,ps.verified_at,ps.settled_at,
|
||
(SELECT count(*)::int FROM payment_attempts attempts WHERE attempts.session_id=ps.id),
|
||
EXISTS(SELECT 1 FROM payment_attempts attempts WHERE attempts.session_id=ps.id AND attempts.status='InBank'),
|
||
last_attempt.status,last_attempt.created_at
|
||
FROM payment_sessions ps JOIN terminals t ON t.id=ps.terminal_id
|
||
LEFT JOIN LATERAL (SELECT status,created_at FROM payment_attempts WHERE session_id=ps.id ORDER BY sequence DESC LIMIT 1) last_attempt ON true
|
||
WHERE ` + detailFilterPredicate + ` ORDER BY ps.created_at DESC,ps.id DESC LIMIT $10`
|
||
rows, err := s.pool.Query(ctx, query, merchantID, period.StartAt, period.EndAt, filter.View, filter.TerminalID, filter.FailureStage, filter.Status, filter.MinAmountRial, filter.MaxAmountRial, MaxExportRows+1)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("querying session export: %w", err)
|
||
}
|
||
defer rows.Close()
|
||
items := make([]ExportDetail, 0, 1024)
|
||
for rows.Next() {
|
||
var id int64
|
||
var reachedBank bool
|
||
var item ExportDetail
|
||
if err := rows.Scan(&id, &item.SessionID, &item.TerminalID, &item.SessionStatus, &item.AmountRial, &item.FeeAmountRial, &item.CreatedAt, &item.ExpiresAt, &item.VerifiedAt, &item.SettledAt, &item.AttemptCount, &reachedBank, &item.LastAttemptStatus, &item.LastAttemptAt); err != nil {
|
||
return nil, fmt.Errorf("scanning session export: %w", err)
|
||
}
|
||
item.SessionID = "ses_" + item.SessionID
|
||
setFailureStage(&item.Detail, reachedBank)
|
||
items = append(items, item)
|
||
if len(items) > MaxExportRows {
|
||
return nil, ErrExportTooLarge
|
||
}
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return nil, fmt.Errorf("iterating session export: %w", err)
|
||
}
|
||
return items, nil
|
||
}
|
||
|
||
func (s *Service) Session(ctx context.Context, merchantID int64, sessionID string) (Detail, bool, error) {
|
||
sourceKey, valid := sessionSourceKey(sessionID)
|
||
if !valid {
|
||
return Detail{}, false, nil
|
||
}
|
||
query := `SELECT ps.id,ps.source_key,t.source_key,ps.status,ps.amount_rial,ps.fee_amount_rial,ps.created_at,ps.expires_at,ps.verified_at,ps.settled_at,
|
||
count(pa.id)::int,coalesce(bool_or(pa.status='InBank'),false) FROM payment_sessions ps JOIN terminals t ON t.id=ps.terminal_id LEFT JOIN payment_attempts pa ON pa.session_id=ps.id
|
||
WHERE ps.merchant_id=$1 AND ps.source_key=$2 GROUP BY ps.id,t.source_key`
|
||
var id int64
|
||
var reachedBank bool
|
||
var detail Detail
|
||
err := s.pool.QueryRow(ctx, query, merchantID, sourceKey).Scan(&id, &detail.SessionID, &detail.TerminalID, &detail.SessionStatus, &detail.AmountRial, &detail.FeeAmountRial, &detail.CreatedAt, &detail.ExpiresAt, &detail.VerifiedAt, &detail.SettledAt, &detail.AttemptCount, &reachedBank)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return Detail{}, false, nil
|
||
}
|
||
if err != nil {
|
||
return Detail{}, false, fmt.Errorf("querying session: %w", err)
|
||
}
|
||
detail.SessionID = "ses_" + detail.SessionID
|
||
detail.Attempts, err = s.attempts(ctx, id)
|
||
if err != nil {
|
||
return Detail{}, false, err
|
||
}
|
||
setFailureStage(&detail, reachedBank)
|
||
return detail, true, nil
|
||
}
|
||
|
||
func (s *Service) attempts(ctx context.Context, sessionID int64) ([]Attempt, error) {
|
||
rows, err := s.pool.Query(ctx, `SELECT sequence,status,created_at,init_time_ms,verify_time_ms FROM payment_attempts WHERE session_id=$1 ORDER BY sequence`, sessionID)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("querying attempts: %w", err)
|
||
}
|
||
attempts, err := pgx.CollectRows(rows, func(row pgx.CollectableRow) (Attempt, error) {
|
||
var attempt Attempt
|
||
err := row.Scan(&attempt.Sequence, &attempt.Status, &attempt.CreatedAt, &attempt.InitTimeMS, &attempt.VerifyTimeMS)
|
||
return attempt, err
|
||
})
|
||
if err != nil {
|
||
return nil, fmt.Errorf("collecting attempts: %w", err)
|
||
}
|
||
return attempts, nil
|
||
}
|
||
|
||
func setFailureStage(detail *Detail, reachedBank bool) {
|
||
if detail.SessionStatus != "Failed" {
|
||
return
|
||
}
|
||
stage := "OTHER_OBSERVED"
|
||
if detail.AttemptCount == 0 {
|
||
stage = "NO_ATTEMPT"
|
||
} else if reachedBank {
|
||
stage = "REACHED_BANK"
|
||
}
|
||
detail.FailureStage = &stage
|
||
}
|
||
|
||
func sessionSourceKey(sessionID string) (string, bool) {
|
||
if !strings.HasPrefix(sessionID, "ses_") || len(sessionID) <= len("ses_") {
|
||
return "", false
|
||
}
|
||
return strings.TrimPrefix(sessionID, "ses_"), true
|
||
}
|
||
|
||
func terminalDisplayName(terminalID string) string {
|
||
suffix := terminalID
|
||
if len(suffix) > 4 {
|
||
suffix = suffix[len(suffix)-4:]
|
||
}
|
||
digits := strings.NewReplacer("0", "۰", "1", "۱", "2", "۲", "3", "۳", "4", "۴", "5", "۵", "6", "۶", "7", "۷", "8", "۸", "9", "۹")
|
||
return "پایانه " + digits.Replace(suffix)
|
||
}
|