wangbo
810dcfeee6
停止持久化 provider 原始响应、兼容响应快照、attempt/event/outbox 重复 JSON，并由标准任务结果动态生成 Kling/Keling/Volces 兼容响应。 增加事件去重与预算、极简 callback 投递、7/30 天分批清理、安全删除条件、并发迁移索引及可实际恢复的任务域排除备份。历史清理默认关闭，待兼容协议和异步恢复在线验证后单独启用。 验证：Go 全量测试与 go vet、PostgreSQL 18 集成与实际备份恢复、迁移安全测试、bash -n、ShellCheck、Compose 配置和人工发布脚本测试均通过。
638 lines
22 KiB
Go
638 lines
22 KiB
Go
package store
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
)
|
|
|
|
const (
|
|
BillingSettlementBatchSize = 50
|
|
BillingSettlementLockTimeout = 120 * time.Second
|
|
BillingSettlementMaxAttempts = 20
|
|
)
|
|
|
|
var ErrBillingSettlementNotRetryable = errors.New("billing settlement is not retryable")
|
|
|
|
type BillingSettlement struct {
|
|
ID string `json:"id"`
|
|
TaskID string `json:"taskId"`
|
|
Action string `json:"action"`
|
|
Amount float64 `json:"amount"`
|
|
Currency string `json:"currency"`
|
|
Status string `json:"status"`
|
|
Attempts int `json:"attempts"`
|
|
NextAttemptAt time.Time `json:"nextAttemptAt"`
|
|
LockedBy string `json:"lockedBy,omitempty"`
|
|
LockedAt string `json:"lockedAt,omitempty"`
|
|
LastErrorCode string `json:"lastErrorCode,omitempty"`
|
|
LastErrorMessage string `json:"lastErrorMessage,omitempty"`
|
|
ManualReviewReason string `json:"manualReviewReason,omitempty"`
|
|
PricingSnapshot map[string]any `json:"pricingSnapshot,omitempty"`
|
|
Payload map[string]any `json:"payload,omitempty"`
|
|
CompletedAt string `json:"completedAt,omitempty"`
|
|
CreatedAt time.Time `json:"createdAt"`
|
|
UpdatedAt time.Time `json:"updatedAt"`
|
|
LockToken string `json:"-"`
|
|
amountExact string
|
|
}
|
|
|
|
type BillingSettlementListFilter struct {
|
|
Status string
|
|
Action string
|
|
Page int
|
|
PageSize int
|
|
}
|
|
|
|
type BillingSettlementListResult struct {
|
|
Items []BillingSettlement `json:"items"`
|
|
Total int `json:"total"`
|
|
Page int `json:"page"`
|
|
PageSize int `json:"pageSize"`
|
|
}
|
|
|
|
type BillingMetricsSnapshot struct {
|
|
SettlementBacklog int64
|
|
SettlementDelaySecs float64
|
|
ManualReview int64
|
|
OrphanFrozen int64
|
|
PricingUnavailable int64
|
|
}
|
|
|
|
func (s *Store) BillingMetrics(ctx context.Context) (BillingMetricsSnapshot, error) {
|
|
var snapshot BillingMetricsSnapshot
|
|
err := s.pool.QueryRow(ctx, `
|
|
WITH open_outbox AS (
|
|
SELECT created_at
|
|
FROM settlement_outbox
|
|
WHERE status IN ('pending', 'processing', 'retryable_failed')
|
|
), active_reservations AS (
|
|
SELECT reserve.account_id, COALESCE(SUM(reserve.amount), 0) AS amount
|
|
FROM gateway_wallet_transactions reserve
|
|
WHERE reserve.transaction_type = 'reserve'
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM gateway_wallet_transactions release
|
|
WHERE release.account_id = reserve.account_id
|
|
AND release.transaction_type = 'release'
|
|
AND release.idempotency_key = reserve.idempotency_key || ':release'
|
|
)
|
|
GROUP BY reserve.account_id
|
|
)
|
|
SELECT
|
|
(SELECT COUNT(*) FROM open_outbox),
|
|
COALESCE((SELECT EXTRACT(EPOCH FROM now() - MIN(created_at)) FROM open_outbox), 0),
|
|
(SELECT COUNT(*) FROM gateway_tasks WHERE billing_status = 'manual_review'),
|
|
(SELECT COUNT(*) FROM gateway_wallet_accounts account
|
|
LEFT JOIN active_reservations reservation ON reservation.account_id = account.id
|
|
WHERE account.frozen_balance <> COALESCE(reservation.amount, 0)),
|
|
(SELECT COUNT(*) FROM gateway_tasks WHERE error_code = 'pricing_unavailable')`).Scan(
|
|
&snapshot.SettlementBacklog,
|
|
&snapshot.SettlementDelaySecs,
|
|
&snapshot.ManualReview,
|
|
&snapshot.OrphanFrozen,
|
|
&snapshot.PricingUnavailable,
|
|
)
|
|
return snapshot, err
|
|
}
|
|
|
|
func (s *Store) ClaimBillingSettlements(ctx context.Context, workerID string, limit int, staleAfter time.Duration) ([]BillingSettlement, error) {
|
|
if limit <= 0 || limit > BillingSettlementBatchSize {
|
|
limit = BillingSettlementBatchSize
|
|
}
|
|
if staleAfter <= 0 {
|
|
staleAfter = BillingSettlementLockTimeout
|
|
}
|
|
rows, err := s.pool.Query(ctx, `
|
|
WITH picked AS (
|
|
SELECT id
|
|
FROM settlement_outbox
|
|
WHERE (
|
|
status IN ('pending', 'retryable_failed')
|
|
AND next_attempt_at <= now()
|
|
)
|
|
OR (
|
|
status = 'processing'
|
|
AND locked_at < now() - ($3::int * interval '1 second')
|
|
)
|
|
ORDER BY next_attempt_at ASC, created_at ASC
|
|
LIMIT $2
|
|
FOR UPDATE SKIP LOCKED
|
|
)
|
|
UPDATE settlement_outbox outbox
|
|
SET status = 'processing',
|
|
attempts = outbox.attempts + 1,
|
|
locked_by = $1,
|
|
lock_token = gen_random_uuid(),
|
|
locked_at = now(),
|
|
updated_at = now()
|
|
FROM picked
|
|
WHERE outbox.id = picked.id
|
|
RETURNING outbox.id::text, outbox.task_id::text, outbox.action, outbox.amount::text,
|
|
outbox.currency, outbox.status, outbox.attempts, outbox.next_attempt_at,
|
|
COALESCE(outbox.locked_by, ''), COALESCE(outbox.lock_token::text, ''),
|
|
COALESCE(outbox.locked_at::text, ''), COALESCE(outbox.last_error_code, ''),
|
|
COALESCE(outbox.last_error_message, ''), COALESCE(outbox.manual_review_reason, ''),
|
|
outbox.pricing_snapshot, outbox.payload, COALESCE(outbox.completed_at::text, ''),
|
|
outbox.created_at, outbox.updated_at`,
|
|
workerID, limit, int(staleAfter/time.Second))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := make([]BillingSettlement, 0)
|
|
for rows.Next() {
|
|
item, err := scanBillingSettlement(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, item)
|
|
}
|
|
return items, rows.Err()
|
|
}
|
|
|
|
func (s *Store) ProcessBillingSettlement(ctx context.Context, settlement BillingSettlement) error {
|
|
return pgx.BeginFunc(ctx, s.pool, func(tx pgx.Tx) error {
|
|
var action string
|
|
var amount string
|
|
var currency string
|
|
var taskID string
|
|
var taskStatus string
|
|
var gatewayUserID string
|
|
var gatewayTenantID string
|
|
var userID string
|
|
var tenantID string
|
|
var tenantKey string
|
|
err := tx.QueryRow(ctx, `
|
|
SELECT outbox.action, outbox.amount::text, outbox.currency, outbox.task_id::text,
|
|
task.status, COALESCE(task.gateway_user_id::text, ''),
|
|
COALESCE(task.gateway_tenant_id::text, ''), task.user_id,
|
|
COALESCE(task.tenant_id, ''), COALESCE(task.tenant_key, '')
|
|
FROM settlement_outbox outbox
|
|
JOIN gateway_tasks task ON task.id = outbox.task_id
|
|
WHERE outbox.id = $1::uuid
|
|
AND outbox.status = 'processing'
|
|
AND outbox.lock_token = $2::uuid
|
|
FOR UPDATE OF outbox, task`, settlement.ID, settlement.LockToken).Scan(
|
|
&action, &amount, ¤cy, &taskID, &taskStatus, &gatewayUserID,
|
|
&gatewayTenantID, &userID, &tenantID, &tenantKey,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if _, err := tx.Exec(ctx, `
|
|
UPDATE gateway_tasks
|
|
SET billing_status = 'processing', billing_updated_at = now(), updated_at = now()
|
|
WHERE id = $1::uuid`, taskID); err != nil {
|
|
return err
|
|
}
|
|
switch action {
|
|
case "settle":
|
|
if taskStatus != "succeeded" {
|
|
return fmt.Errorf("settle action requires succeeded task")
|
|
}
|
|
if err := settleBillingOutboxTx(ctx, tx, taskID, gatewayUserID, gatewayTenantID, userID, tenantID, tenantKey, currency, amount); err != nil {
|
|
return err
|
|
}
|
|
return completeBillingOutboxTx(ctx, tx, settlement.ID, settlement.LockToken, taskID, "settled")
|
|
case "release":
|
|
if err := releaseBillingOutboxTx(ctx, tx, taskID, gatewayUserID, gatewayTenantID, currency); err != nil {
|
|
return err
|
|
}
|
|
return completeBillingOutboxTx(ctx, tx, settlement.ID, settlement.LockToken, taskID, "released")
|
|
default:
|
|
return fmt.Errorf("unsupported billing settlement action %q", action)
|
|
}
|
|
})
|
|
}
|
|
|
|
func settleBillingOutboxTx(
|
|
ctx context.Context,
|
|
tx pgx.Tx,
|
|
taskID string,
|
|
gatewayUserID string,
|
|
gatewayTenantID string,
|
|
userID string,
|
|
tenantID string,
|
|
tenantKey string,
|
|
currency string,
|
|
amount string,
|
|
) error {
|
|
if gatewayUserID == "" {
|
|
return fmt.Errorf("task %s has no gateway wallet user", taskID)
|
|
}
|
|
if _, err := tx.Exec(ctx, `
|
|
INSERT INTO gateway_wallet_accounts (
|
|
gateway_tenant_id, gateway_user_id, tenant_id, tenant_key, user_id, currency
|
|
)
|
|
VALUES (NULLIF($1, '')::uuid, $2::uuid, NULLIF($3, ''), NULLIF($4, ''), NULLIF($5, ''), $6)
|
|
ON CONFLICT (gateway_user_id, currency) DO NOTHING`,
|
|
gatewayTenantID, gatewayUserID, tenantID, tenantKey, userID, currency); err != nil {
|
|
return err
|
|
}
|
|
if err := ensureWalletAccountAuditGuard(ctx, tx, gatewayUserID, currency); err != nil {
|
|
return err
|
|
}
|
|
var accountID string
|
|
var balanceBefore string
|
|
var frozenBefore string
|
|
if err := tx.QueryRow(ctx, `
|
|
SELECT id::text, balance::text, frozen_balance::text
|
|
FROM gateway_wallet_accounts
|
|
WHERE gateway_user_id = $1::uuid AND currency = $2
|
|
FOR UPDATE`, gatewayUserID, currency).Scan(&accountID, &balanceBefore, &frozenBefore); err != nil {
|
|
return err
|
|
}
|
|
var alreadySettled bool
|
|
if err := tx.QueryRow(ctx, `
|
|
SELECT EXISTS (
|
|
SELECT 1 FROM gateway_wallet_transactions
|
|
WHERE account_id = $1::uuid AND idempotency_key = $2
|
|
)`, accountID, billingIdempotencyKey(taskID)).Scan(&alreadySettled); err != nil {
|
|
return err
|
|
}
|
|
if alreadySettled {
|
|
return nil
|
|
}
|
|
|
|
var reservationKey string
|
|
var reservedAmount string
|
|
if err := tx.QueryRow(ctx, `
|
|
SELECT COALESCE(reserve.idempotency_key, ''), reserve.amount::text
|
|
FROM gateway_wallet_transactions reserve
|
|
WHERE reserve.account_id = $1::uuid
|
|
AND reserve.reference_type = 'gateway_task'
|
|
AND reserve.reference_id = $2
|
|
AND reserve.transaction_type = 'reserve'
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM gateway_wallet_transactions release
|
|
WHERE release.account_id = reserve.account_id
|
|
AND release.transaction_type = 'release'
|
|
AND release.idempotency_key = reserve.idempotency_key || ':release'
|
|
)
|
|
ORDER BY reserve.created_at DESC
|
|
LIMIT 1`, accountID, taskID).Scan(&reservationKey, &reservedAmount); errors.Is(err, pgx.ErrNoRows) {
|
|
reservationKey = ""
|
|
reservedAmount = "0"
|
|
} else if err != nil {
|
|
return err
|
|
}
|
|
|
|
var balanceAfter string
|
|
var frozenAfter string
|
|
err := tx.QueryRow(ctx, `
|
|
UPDATE gateway_wallet_accounts
|
|
SET balance = balance - $2::numeric,
|
|
total_spent = total_spent + $2::numeric,
|
|
frozen_balance = GREATEST(0, frozen_balance - $3::numeric),
|
|
updated_at = now()
|
|
WHERE id = $1::uuid
|
|
AND balance - frozen_balance + $3::numeric >= $2::numeric
|
|
RETURNING balance::text, frozen_balance::text`,
|
|
accountID, amount, reservedAmount).Scan(&balanceAfter, &frozenAfter)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return fmt.Errorf("%w: task %s settlement amount exceeds spendable balance", ErrInsufficientWalletBalance, taskID)
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
var reservedPositive bool
|
|
if err := tx.QueryRow(ctx, `SELECT $1::numeric > 0`, reservedAmount).Scan(&reservedPositive); err != nil {
|
|
return err
|
|
}
|
|
if reservedPositive {
|
|
releaseMetadata, _ := json.Marshal(map[string]any{
|
|
"taskId": taskID, "reason": "task_billing_settled",
|
|
"reserved": reservedAmount, "frozenBefore": frozenBefore, "frozenAfter": frozenAfter,
|
|
})
|
|
if _, err := tx.Exec(ctx, `
|
|
INSERT INTO gateway_wallet_transactions (
|
|
account_id, gateway_tenant_id, gateway_user_id, direction, transaction_type,
|
|
amount, balance_before, balance_after, idempotency_key, reference_type, reference_id, metadata
|
|
)
|
|
VALUES (
|
|
$1::uuid, NULLIF($2, '')::uuid, $3::uuid, 'credit', 'release',
|
|
$4::numeric, $5::numeric, $5::numeric, $6, 'gateway_task', $7, $8::jsonb
|
|
)
|
|
ON CONFLICT (account_id, idempotency_key) WHERE idempotency_key IS NOT NULL DO NOTHING`,
|
|
accountID, gatewayTenantID, gatewayUserID, reservedAmount, balanceBefore,
|
|
billingReservationReleaseIdempotencyKey(reservationKey), taskID, string(releaseMetadata)); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
metadata, _ := json.Marshal(map[string]any{
|
|
"taskId": taskID, "reservedAmount": reservedAmount,
|
|
"frozenBefore": frozenBefore, "frozenAfter": frozenAfter,
|
|
})
|
|
_, err = tx.Exec(ctx, `
|
|
INSERT INTO gateway_wallet_transactions (
|
|
account_id, gateway_tenant_id, gateway_user_id, direction, transaction_type,
|
|
amount, balance_before, balance_after, idempotency_key, reference_type, reference_id, metadata
|
|
)
|
|
VALUES (
|
|
$1::uuid, NULLIF($2, '')::uuid, $3::uuid, 'debit', 'task_billing',
|
|
$4::numeric, $5::numeric, $6::numeric, $7, 'gateway_task', $8, $9::jsonb
|
|
)
|
|
ON CONFLICT (account_id, idempotency_key) WHERE idempotency_key IS NOT NULL DO NOTHING`,
|
|
accountID, gatewayTenantID, gatewayUserID, amount, balanceBefore, balanceAfter,
|
|
billingIdempotencyKey(taskID), taskID, string(metadata))
|
|
return err
|
|
}
|
|
|
|
func releaseBillingOutboxTx(ctx context.Context, tx pgx.Tx, taskID string, gatewayUserID string, gatewayTenantID string, currency string) error {
|
|
if gatewayUserID == "" {
|
|
return nil
|
|
}
|
|
var accountID string
|
|
var balance string
|
|
var frozenBefore string
|
|
err := tx.QueryRow(ctx, `
|
|
SELECT id::text, balance::text, frozen_balance::text
|
|
FROM gateway_wallet_accounts
|
|
WHERE gateway_user_id = $1::uuid AND currency = $2
|
|
FOR UPDATE`, gatewayUserID, currency).Scan(&accountID, &balance, &frozenBefore)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var reservationKey string
|
|
var reservedAmount string
|
|
err = tx.QueryRow(ctx, `
|
|
SELECT COALESCE(reserve.idempotency_key, ''), reserve.amount::text
|
|
FROM gateway_wallet_transactions reserve
|
|
WHERE reserve.account_id = $1::uuid
|
|
AND reserve.reference_type = 'gateway_task'
|
|
AND reserve.reference_id = $2
|
|
AND reserve.transaction_type = 'reserve'
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM gateway_wallet_transactions release
|
|
WHERE release.account_id = reserve.account_id
|
|
AND release.transaction_type = 'release'
|
|
AND release.idempotency_key = reserve.idempotency_key || ':release'
|
|
)
|
|
ORDER BY reserve.created_at DESC
|
|
LIMIT 1`, accountID, taskID).Scan(&reservationKey, &reservedAmount)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var frozenAfter string
|
|
if err := tx.QueryRow(ctx, `
|
|
UPDATE gateway_wallet_accounts
|
|
SET frozen_balance = frozen_balance - $2::numeric,
|
|
updated_at = now()
|
|
WHERE id = $1::uuid
|
|
AND frozen_balance >= $2::numeric
|
|
RETURNING frozen_balance::text`, accountID, reservedAmount).Scan(&frozenAfter); err != nil {
|
|
return err
|
|
}
|
|
metadata, _ := json.Marshal(map[string]any{
|
|
"taskId": taskID, "reason": "task_terminal_release",
|
|
"reserved": reservedAmount, "frozenBefore": frozenBefore, "frozenAfter": frozenAfter,
|
|
})
|
|
_, err = tx.Exec(ctx, `
|
|
INSERT INTO gateway_wallet_transactions (
|
|
account_id, gateway_tenant_id, gateway_user_id, direction, transaction_type,
|
|
amount, balance_before, balance_after, idempotency_key, reference_type, reference_id, metadata
|
|
)
|
|
VALUES (
|
|
$1::uuid, NULLIF($2, '')::uuid, $3::uuid, 'credit', 'release',
|
|
$4::numeric, $5::numeric, $5::numeric, $6, 'gateway_task', $7, $8::jsonb
|
|
)
|
|
ON CONFLICT (account_id, idempotency_key) WHERE idempotency_key IS NOT NULL DO NOTHING`,
|
|
accountID, gatewayTenantID, gatewayUserID, reservedAmount, balance,
|
|
billingReservationReleaseIdempotencyKey(reservationKey), taskID, string(metadata))
|
|
return err
|
|
}
|
|
|
|
func completeBillingOutboxTx(ctx context.Context, tx pgx.Tx, settlementID string, lockToken string, taskID string, billingStatus string) error {
|
|
if _, err := tx.Exec(ctx, `
|
|
UPDATE gateway_tasks
|
|
SET billing_status = $2,
|
|
reservation_amount = 0,
|
|
billing_updated_at = now(),
|
|
billing_settled_at = CASE WHEN $2 = 'settled' THEN now() ELSE billing_settled_at END,
|
|
updated_at = now()
|
|
WHERE id = $1::uuid`, taskID, billingStatus); err != nil {
|
|
return err
|
|
}
|
|
if _, err := tx.Exec(ctx, `
|
|
INSERT INTO gateway_task_events (task_id, seq, event_type, status, phase, progress, message, payload, simulated)
|
|
SELECT task.id,
|
|
COALESCE((SELECT MAX(event.seq) + 1 FROM gateway_task_events event WHERE event.task_id = task.id), 1),
|
|
CASE WHEN $2 = 'settled' THEN 'task.billing.settled' ELSE 'task.billing.released' END,
|
|
task.status,
|
|
NULL,
|
|
0,
|
|
NULL,
|
|
'{}'::jsonb,
|
|
task.run_mode = 'simulation'
|
|
FROM gateway_tasks task
|
|
WHERE task.id = $1::uuid`, taskID, billingStatus); err != nil {
|
|
return err
|
|
}
|
|
tag, err := tx.Exec(ctx, `
|
|
UPDATE settlement_outbox
|
|
SET status = 'completed',
|
|
completed_at = now(),
|
|
locked_by = NULL,
|
|
lock_token = NULL,
|
|
locked_at = NULL,
|
|
last_error_code = NULL,
|
|
last_error_message = NULL,
|
|
updated_at = now()
|
|
WHERE id = $1::uuid
|
|
AND status = 'processing'
|
|
AND lock_token = $2::uuid`, settlementID, lockToken)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if tag.RowsAffected() != 1 {
|
|
return pgx.ErrNoRows
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Store) MarkBillingSettlementFailed(ctx context.Context, settlement BillingSettlement, code string, message string, delay time.Duration) error {
|
|
if delay < time.Second {
|
|
delay = time.Second
|
|
}
|
|
if delay > 15*time.Minute {
|
|
delay = 15 * time.Minute
|
|
}
|
|
manualReview := settlement.Attempts >= BillingSettlementMaxAttempts
|
|
status := "retryable_failed"
|
|
taskStatus := "retryable_failed"
|
|
manualReason := ""
|
|
if manualReview {
|
|
status = "manual_review"
|
|
taskStatus = "manual_review"
|
|
manualReason = "maximum settlement attempts exceeded"
|
|
}
|
|
return pgx.BeginFunc(ctx, s.pool, func(tx pgx.Tx) error {
|
|
tag, err := tx.Exec(ctx, `
|
|
UPDATE settlement_outbox
|
|
SET status = $3,
|
|
next_attempt_at = now() + ($4::int * interval '1 second'),
|
|
locked_by = NULL,
|
|
lock_token = NULL,
|
|
locked_at = NULL,
|
|
last_error_code = NULLIF($5, ''),
|
|
last_error_message = NULLIF($6, ''),
|
|
manual_review_reason = NULLIF($7, ''),
|
|
updated_at = now()
|
|
WHERE id = $1::uuid
|
|
AND lock_token = $2::uuid
|
|
AND status = 'processing'`,
|
|
settlement.ID, settlement.LockToken, status, int(delay/time.Second), code, message, manualReason)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if tag.RowsAffected() != 1 {
|
|
return pgx.ErrNoRows
|
|
}
|
|
_, err = tx.Exec(ctx, `
|
|
UPDATE gateway_tasks
|
|
SET billing_status = $2,
|
|
billing_updated_at = now(),
|
|
updated_at = now()
|
|
WHERE id = $1::uuid`, settlement.TaskID, taskStatus)
|
|
return err
|
|
})
|
|
}
|
|
|
|
func (s *Store) ListBillingSettlements(ctx context.Context, filter BillingSettlementListFilter) (BillingSettlementListResult, error) {
|
|
page := filter.Page
|
|
if page <= 0 {
|
|
page = 1
|
|
}
|
|
pageSize := filter.PageSize
|
|
if pageSize <= 0 {
|
|
pageSize = 50
|
|
}
|
|
if pageSize > 100 {
|
|
pageSize = 100
|
|
}
|
|
status := strings.TrimSpace(filter.Status)
|
|
action := strings.TrimSpace(filter.Action)
|
|
var total int
|
|
if err := s.pool.QueryRow(ctx, `
|
|
SELECT count(*)
|
|
FROM settlement_outbox
|
|
WHERE (NULLIF($1, '') IS NULL OR status = $1)
|
|
AND (NULLIF($2, '') IS NULL OR action = $2)`, status, action).Scan(&total); err != nil {
|
|
return BillingSettlementListResult{}, err
|
|
}
|
|
rows, err := s.pool.Query(ctx, `
|
|
SELECT id::text, task_id::text, action, amount::text, currency, status, attempts,
|
|
next_attempt_at, COALESCE(locked_by, ''), COALESCE(lock_token::text, ''),
|
|
COALESCE(locked_at::text, ''), COALESCE(last_error_code, ''),
|
|
COALESCE(last_error_message, ''), COALESCE(manual_review_reason, ''),
|
|
pricing_snapshot, payload, COALESCE(completed_at::text, ''), created_at, updated_at
|
|
FROM settlement_outbox
|
|
WHERE (NULLIF($1, '') IS NULL OR status = $1)
|
|
AND (NULLIF($2, '') IS NULL OR action = $2)
|
|
ORDER BY created_at DESC
|
|
LIMIT $3 OFFSET $4`, status, action, pageSize, (page-1)*pageSize)
|
|
if err != nil {
|
|
return BillingSettlementListResult{}, err
|
|
}
|
|
defer rows.Close()
|
|
items := make([]BillingSettlement, 0)
|
|
for rows.Next() {
|
|
item, err := scanBillingSettlement(rows)
|
|
if err != nil {
|
|
return BillingSettlementListResult{}, err
|
|
}
|
|
items = append(items, item)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return BillingSettlementListResult{}, err
|
|
}
|
|
return BillingSettlementListResult{Items: items, Total: total, Page: page, PageSize: pageSize}, nil
|
|
}
|
|
|
|
func (s *Store) RetryBillingSettlementTx(ctx context.Context, tx Tx, id string, idempotencyKeyHash string) (BillingSettlement, bool, error) {
|
|
current, err := scanBillingSettlement(tx.QueryRow(ctx, `
|
|
SELECT id::text, task_id::text, action, amount::text, currency, status, attempts,
|
|
next_attempt_at, COALESCE(locked_by, ''), COALESCE(lock_token::text, ''),
|
|
COALESCE(locked_at::text, ''), COALESCE(last_error_code, ''),
|
|
COALESCE(last_error_message, ''), COALESCE(manual_review_reason, ''),
|
|
pricing_snapshot, payload, COALESCE(completed_at::text, ''), created_at, updated_at
|
|
FROM settlement_outbox
|
|
WHERE id = $1::uuid
|
|
FOR UPDATE`, id))
|
|
if err != nil {
|
|
return BillingSettlement{}, false, err
|
|
}
|
|
var recordedHash string
|
|
if err := tx.QueryRow(ctx, `
|
|
SELECT COALESCE(retry_idempotency_key_hash, '')
|
|
FROM settlement_outbox
|
|
WHERE id = $1::uuid`, id).Scan(&recordedHash); err != nil {
|
|
return BillingSettlement{}, false, err
|
|
}
|
|
if recordedHash != "" && recordedHash == idempotencyKeyHash {
|
|
return current, true, nil
|
|
}
|
|
if current.Status != "retryable_failed" && current.Status != "manual_review" {
|
|
return BillingSettlement{}, false, ErrBillingSettlementNotRetryable
|
|
}
|
|
item, err := scanBillingSettlement(tx.QueryRow(ctx, `
|
|
UPDATE settlement_outbox
|
|
SET status = 'pending',
|
|
next_attempt_at = now(),
|
|
locked_by = NULL,
|
|
lock_token = NULL,
|
|
locked_at = NULL,
|
|
manual_review_reason = NULL,
|
|
retry_idempotency_key_hash = $2,
|
|
retry_requested_at = now(),
|
|
updated_at = now()
|
|
WHERE id = $1::uuid
|
|
AND status IN ('retryable_failed', 'manual_review')
|
|
RETURNING id::text, task_id::text, action, amount::text, currency, status, attempts,
|
|
next_attempt_at, COALESCE(locked_by, ''), COALESCE(lock_token::text, ''),
|
|
COALESCE(locked_at::text, ''), COALESCE(last_error_code, ''),
|
|
COALESCE(last_error_message, ''), COALESCE(manual_review_reason, ''),
|
|
pricing_snapshot, payload, COALESCE(completed_at::text, ''), created_at, updated_at`, id, idempotencyKeyHash))
|
|
if err != nil {
|
|
return BillingSettlement{}, false, err
|
|
}
|
|
if _, err := tx.Exec(ctx, `
|
|
UPDATE gateway_tasks
|
|
SET billing_status = 'pending', billing_updated_at = now(), updated_at = now()
|
|
WHERE id = $1::uuid`, item.TaskID); err != nil {
|
|
return BillingSettlement{}, false, err
|
|
}
|
|
return item, false, nil
|
|
}
|
|
|
|
func scanBillingSettlement(scanner taskScanner) (BillingSettlement, error) {
|
|
var item BillingSettlement
|
|
var pricingSnapshotBytes []byte
|
|
var payloadBytes []byte
|
|
if err := scanner.Scan(
|
|
&item.ID, &item.TaskID, &item.Action, &item.amountExact, &item.Currency,
|
|
&item.Status, &item.Attempts, &item.NextAttemptAt, &item.LockedBy, &item.LockToken,
|
|
&item.LockedAt, &item.LastErrorCode, &item.LastErrorMessage, &item.ManualReviewReason,
|
|
&pricingSnapshotBytes, &payloadBytes, &item.CompletedAt, &item.CreatedAt, &item.UpdatedAt,
|
|
); err != nil {
|
|
return BillingSettlement{}, err
|
|
}
|
|
item.Amount, _ = strconv.ParseFloat(item.amountExact, 64)
|
|
item.PricingSnapshot = decodeObject(pricingSnapshotBytes)
|
|
item.Payload = decodeObject(payloadBytes)
|
|
return item, nil
|
|
}
|