package store

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"hash/fnv"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/easyai/easyai-ai-gateway/apps/api/internal/auth"
	"github.com/jackc/pgx/v5"
)

const walletReservationLockStripes = 256

type walletReservationLockSet struct {
	stripes [walletReservationLockStripes]sync.Mutex
}

func (locks *walletReservationLockSet) lock(gatewayUserID string) func() {
	hasher := fnv.New32a()
	_, _ = hasher.Write([]byte(strings.TrimSpace(gatewayUserID)))
	stripe := &locks.stripes[hasher.Sum32()%walletReservationLockStripes]
	stripe.Lock()
	return stripe.Unlock
}

type GatewayWalletAccount struct {
	ID              string         `json:"id"`
	GatewayTenantID string         `json:"gatewayTenantId,omitempty"`
	GatewayUserID   string         `json:"gatewayUserId"`
	TenantID        string         `json:"tenantId,omitempty"`
	TenantKey       string         `json:"tenantKey,omitempty"`
	UserID          string         `json:"userId,omitempty"`
	Currency        string         `json:"currency"`
	Balance         float64        `json:"balance"`
	FrozenBalance   float64        `json:"frozenBalance"`
	TotalRecharged  float64        `json:"totalRecharged"`
	TotalSpent      float64        `json:"totalSpent"`
	Status          string         `json:"status"`
	Metadata        map[string]any `json:"metadata,omitempty"`
	CreatedAt       time.Time      `json:"createdAt"`
	UpdatedAt       time.Time      `json:"updatedAt"`
}

type GatewayWalletTransaction struct {
	ID              string         `json:"id"`
	AccountID       string         `json:"accountId"`
	Currency        string         `json:"currency,omitempty"`
	GatewayTenantID string         `json:"gatewayTenantId,omitempty"`
	GatewayUserID   string         `json:"gatewayUserId,omitempty"`
	Direction       string         `json:"direction"`
	TransactionType string         `json:"transactionType"`
	Amount          float64        `json:"amount"`
	BalanceBefore   float64        `json:"balanceBefore"`
	BalanceAfter    float64        `json:"balanceAfter"`
	IdempotencyKey  string         `json:"idempotencyKey,omitempty"`
	ReferenceType   string         `json:"referenceType,omitempty"`
	ReferenceID     string         `json:"referenceId,omitempty"`
	Metadata        map[string]any `json:"metadata,omitempty"`
	CreatedAt       time.Time      `json:"createdAt"`
}

type WalletAvailability struct {
	Account         GatewayWalletAccount `json:"account"`
	Currency        string               `json:"currency"`
	RequiredAmount  float64              `json:"requiredAmount"`
	AvailableAmount float64              `json:"availableAmount"`
	Enough          bool                 `json:"enough"`
}

type WalletSummary struct {
	Accounts       []GatewayWalletAccount `json:"accounts"`
	PrimaryAccount GatewayWalletAccount   `json:"primaryAccount"`
}

type WalletTransactionListFilter struct {
	Query           string
	Direction       string
	TransactionType string
	CreatedFrom     *time.Time
	CreatedTo       *time.Time
	Page            int
	PageSize        int
}

type WalletTransactionListResult struct {
	Items    []GatewayWalletTransaction
	Total    int
	Page     int
	PageSize int
}

type WalletBalanceAdjustmentInput struct {
	GatewayUserID  string         `json:"gatewayUserId"`
	Currency       string         `json:"currency"`
	Balance        float64        `json:"balance"`
	BalanceText    string         `json:"-"`
	Reason         string         `json:"reason"`
	IdempotencyKey string         `json:"idempotencyKey"`
	Metadata       map[string]any `json:"metadata"`
}

type WalletRechargeInput struct {
	GatewayUserID  string         `json:"gatewayUserId"`
	Currency       string         `json:"currency"`
	Amount         float64        `json:"amount"`
	AmountText     string         `json:"-"`
	Reason         string         `json:"reason"`
	IdempotencyKey string         `json:"idempotencyKey"`
	Metadata       map[string]any `json:"metadata"`
}

type WalletAdjustmentResult struct {
	Account     GatewayWalletAccount     `json:"account"`
	Before      GatewayWalletAccount     `json:"before"`
	Transaction GatewayWalletTransaction `json:"transaction"`
}

type WalletBillingReservation struct {
	TaskID          string  `json:"taskId"`
	AccountID       string  `json:"accountId"`
	GatewayUserID   string  `json:"gatewayUserId"`
	GatewayTenantID string  `json:"gatewayTenantId,omitempty"`
	Currency        string  `json:"currency"`
	Amount          float64 `json:"amount"`
	IdempotencyKey  string  `json:"idempotencyKey"`
}

func (s *Store) WalletAvailability(ctx context.Context, user *auth.User, currency string, requiredAmount float64) (WalletAvailability, error) {
	gatewayUserID := localGatewayUserID(user)
	if gatewayUserID == "" {
		return WalletAvailability{Currency: normalizeWalletCurrency(currency), RequiredAmount: requiredAmount, Enough: true}, nil
	}
	account, err := s.ensureWalletAccount(ctx, s.pool, gatewayUserID, currency)
	if err != nil {
		return WalletAvailability{}, err
	}
	available := roundMoney(account.Balance - account.FrozenBalance)
	result := WalletAvailability{
		Account:         account,
		Currency:        account.Currency,
		RequiredAmount:  roundMoney(requiredAmount),
		AvailableAmount: available,
		Enough:          available+0.000001 >= requiredAmount,
	}
	if !result.Enough {
		return result, fmt.Errorf("%w: required %.6f %s, available %.6f", ErrInsufficientWalletBalance, requiredAmount, account.Currency, available)
	}
	return result, nil
}

func (s *Store) ReserveTaskBilling(ctx context.Context, task GatewayTask, user *auth.User, billings []any, pricingSnapshots ...map[string]any) ([]WalletBillingReservation, error) {
	gatewayUserID := taskGatewayUserID(task, user)
	if gatewayUserID == "" {
		return nil, nil
	}
	taskID := strings.TrimSpace(task.ID)
	if taskID == "" {
		return nil, fmt.Errorf("task id is required for wallet reservation")
	}

	pricingSnapshot := map[string]any{}
	if len(pricingSnapshots) > 0 {
		pricingSnapshot = emptyObjectIfNil(pricingSnapshots[0])
	}
	if exactAmount := walletString(pricingSnapshot["reservationAmount"]); exactAmount != "" {
		unlock := s.walletReservationLocks.lock(gatewayUserID)
		defer unlock()
		return s.reserveTaskBillingExact(ctx, task, gatewayUserID, exactAmount, pricingSnapshot)
	}
	amounts := walletBillingAmounts(billings)
	if len(amounts) == 0 {
		return nil, nil
	}
	unlock := s.walletReservationLocks.lock(gatewayUserID)
	defer unlock()

	reservations := make([]WalletBillingReservation, 0, len(amounts))
	pricingSnapshotJSON, _ := json.Marshal(sanitizeJSONForStorage(pricingSnapshot))
	requestFingerprint := walletString(pricingSnapshot["requestFingerprint"])
	err := s.beginTransaction(ctx, func(tx pgx.Tx) error {
		for currency, rawAmount := range amounts {
			amount := roundMoney(rawAmount)
			if amount <= 0 {
				continue
			}

			account, err := s.ensureWalletAccount(ctx, tx, gatewayUserID, currency)
			if err != nil {
				return err
			}

			locked, err := lockWalletAccount(ctx, tx, account.ID)
			if err != nil {
				return err
			}
			activeKey, activeAmount, err := activeWalletReservation(ctx, tx, locked.ID, taskID)
			if err != nil {
				return err
			}
			if activeAmount > 0 {
				reservation := WalletBillingReservation{
					TaskID:          taskID,
					AccountID:       locked.ID,
					GatewayUserID:   gatewayUserID,
					GatewayTenantID: firstNonEmpty(locked.GatewayTenantID, task.GatewayTenantID),
					Currency:        locked.Currency,
					Amount:          activeAmount,
					IdempotencyKey:  activeKey,
				}
				reservations = append(reservations, reservation)
				continue
			}

			sequence, err := nextWalletReservationSequence(ctx, tx, locked.ID, taskID)
			if err != nil {
				return err
			}
			key := billingReservationIdempotencyKey(taskID, locked.Currency, sequence)
			reservation := WalletBillingReservation{
				TaskID:          taskID,
				AccountID:       locked.ID,
				GatewayUserID:   gatewayUserID,
				GatewayTenantID: firstNonEmpty(locked.GatewayTenantID, task.GatewayTenantID),
				Currency:        locked.Currency,
				Amount:          amount,
				IdempotencyKey:  key,
			}
			available := roundMoney(locked.Balance - locked.FrozenBalance)
			if available+0.000001 < amount {
				return fmt.Errorf("%w: required %.6f %s, available %.6f", ErrInsufficientWalletBalance, amount, locked.Currency, available)
			}

			frozenAfter := roundMoney(locked.FrozenBalance + amount)
			if _, err := tx.Exec(ctx, `
UPDATE gateway_wallet_accounts
SET frozen_balance = $2,
    updated_at = now()
WHERE id = $1::uuid`, locked.ID, frozenAfter); err != nil {
				return err
			}
			modelIdentity := taskBillingModelIdentity(GatewayTask{
				Kind:           task.Kind,
				Model:          task.Model,
				RequestedModel: task.RequestedModel,
				ResolvedModel:  task.ResolvedModel,
				Billings:       billings,
				Metrics:        task.Metrics,
			})
			metadata, _ := json.Marshal(map[string]any{
				"taskId":         taskID,
				"kind":           task.Kind,
				"model":          modelIdentity.Model,
				"requestedModel": modelIdentity.RequestedModel,
				"resolvedModel":  modelIdentity.ResolvedModel,
				"modelName":      modelIdentity.ModelName,
				"modelAlias":     modelIdentity.ModelAlias,
				"providerModel":  modelIdentity.ProviderModelName,
				"reserved":       amount,
				"balance":        roundMoney(locked.Balance),
				"frozenBefore":   roundMoney(locked.FrozenBalance),
				"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, 'debit', 'reserve',
  $4, $5, $6, $7, 'gateway_task', $8, $9::jsonb
)`,
				locked.ID,
				firstNonEmpty(locked.GatewayTenantID, task.GatewayTenantID),
				gatewayUserID,
				amount,
				roundMoney(locked.Balance),
				roundMoney(locked.Balance),
				key,
				taskID,
				string(metadata),
			); err != nil {
				return err
			}
			reservations = append(reservations, reservation)
		}
		_, err := tx.Exec(ctx, `
UPDATE gateway_tasks task
SET reservation_amount = COALESCE((
      SELECT SUM(reserve.amount)
      FROM gateway_wallet_transactions reserve
      WHERE reserve.reference_type = 'gateway_task'
        AND reserve.reference_id = task.id::text
        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'
        )
    ), 0),
    billing_status = 'pending',
    billing_currency = 'resource',
    pricing_snapshot = $2::jsonb,
    request_fingerprint = NULLIF($3, ''),
    billing_updated_at = now(),
    updated_at = now()
WHERE task.id = $1::uuid`, taskID, string(pricingSnapshotJSON), requestFingerprint)
		return err
	})
	if err != nil {
		return nil, err
	}
	return reservations, err
}

func (s *Store) reserveTaskBillingExact(ctx context.Context, task GatewayTask, gatewayUserID string, amount string, pricingSnapshot map[string]any) ([]WalletBillingReservation, error) {
	taskID := strings.TrimSpace(task.ID)
	currency := normalizeWalletCurrency(walletString(pricingSnapshot["currency"]))
	if currency != "resource" {
		return nil, fmt.Errorf("unsupported billing currency %q", currency)
	}
	pricingSnapshotJSON, _ := json.Marshal(sanitizeJSONForStorage(pricingSnapshot))
	requestFingerprint := walletString(pricingSnapshot["requestFingerprint"])
	var reservations []WalletBillingReservation
	err := s.beginTransaction(ctx, func(tx pgx.Tx) error {
		var positive bool
		if err := tx.QueryRow(ctx, `SELECT $1::numeric(38, 9) > 0`, amount).Scan(&positive); err != nil {
			return fmt.Errorf("invalid exact reservation amount: %w", err)
		}
		if !positive {
			return nil
		}
		account, err := s.ensureWalletAccount(ctx, tx, gatewayUserID, currency)
		if err != nil {
			return err
		}
		locked, err := lockWalletAccount(ctx, tx, account.ID)
		if err != nil {
			return err
		}
		activeKey, activeAmount, err := activeWalletReservationExact(ctx, tx, locked.ID, taskID)
		if err != nil {
			return err
		}
		if activeAmount != "" {
			presentationAmount, _ := strconv.ParseFloat(activeAmount, 64)
			reservations = []WalletBillingReservation{{
				TaskID: taskID, AccountID: locked.ID, GatewayUserID: gatewayUserID,
				GatewayTenantID: firstNonEmpty(locked.GatewayTenantID, task.GatewayTenantID),
				Currency:        locked.Currency, Amount: presentationAmount, IdempotencyKey: activeKey,
			}}
			return nil
		}
		sequence, err := nextWalletReservationSequence(ctx, tx, locked.ID, taskID)
		if err != nil {
			return err
		}
		key := billingReservationIdempotencyKey(taskID, locked.Currency, sequence)
		var balanceBefore string
		var frozenBefore string
		if err := tx.QueryRow(ctx, `
SELECT balance::text, frozen_balance::text
FROM gateway_wallet_accounts
WHERE id = $1::uuid`, locked.ID).Scan(&balanceBefore, &frozenBefore); err != nil {
			return err
		}
		var frozenAfter string
		err = tx.QueryRow(ctx, `
UPDATE gateway_wallet_accounts
SET frozen_balance = frozen_balance + $2::numeric(38, 9), updated_at = now()
WHERE id = $1::uuid
  AND balance - frozen_balance >= $2::numeric(38, 9)
RETURNING frozen_balance::text`, locked.ID, amount).Scan(&frozenAfter)
		if errors.Is(err, pgx.ErrNoRows) {
			return ErrInsufficientWalletBalance
		}
		if err != nil {
			return err
		}
		metadata, _ := json.Marshal(map[string]any{
			"taskId": taskID, "reserved": amount, "balance": balanceBefore,
			"frozenBefore": frozenBefore, "frozenAfter": frozenAfter,
			"pricingVersion": walletString(pricingSnapshot["pricingVersion"]),
		})
		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, 'debit', 'reserve',
  $4::numeric(38, 9), $5::numeric, $5::numeric, $6, 'gateway_task', $7, $8::jsonb
)`, locked.ID, firstNonEmpty(locked.GatewayTenantID, task.GatewayTenantID), gatewayUserID,
			amount, balanceBefore, key, taskID, string(metadata)); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `
UPDATE gateway_tasks
SET reservation_amount = $2::numeric(38, 9), billing_status = 'pending',
    billing_currency = $3, pricing_snapshot = $4::jsonb,
    request_fingerprint = NULLIF($5, ''), billing_updated_at = now(), updated_at = now()
WHERE id = $1::uuid`, taskID, amount, currency, string(pricingSnapshotJSON), requestFingerprint); err != nil {
			return err
		}
		presentationAmount, _ := strconv.ParseFloat(amount, 64)
		reservations = []WalletBillingReservation{{
			TaskID: taskID, AccountID: locked.ID, GatewayUserID: gatewayUserID,
			GatewayTenantID: firstNonEmpty(locked.GatewayTenantID, task.GatewayTenantID),
			Currency:        locked.Currency, Amount: presentationAmount, IdempotencyKey: key,
		}}
		return nil
	})
	return reservations, err
}

func (s *Store) ReleaseTaskBillingReservations(ctx context.Context, reservations []WalletBillingReservation, reason string) error {
	if len(reservations) == 0 {
		return nil
	}
	reason = strings.TrimSpace(reason)
	if reason == "" {
		reason = "task_not_settled"
	}
	return s.beginTransaction(ctx, func(tx pgx.Tx) error {
		taskIDs := map[string]struct{}{}
		for _, reservation := range reservations {
			if strings.TrimSpace(reservation.AccountID) == "" {
				continue
			}
			if taskID := strings.TrimSpace(reservation.TaskID); taskID != "" {
				taskIDs[taskID] = struct{}{}
			}
			reserveKey := strings.TrimSpace(reservation.IdempotencyKey)
			if reserveKey == "" {
				reserveKey = billingReservationIdempotencyKey(reservation.TaskID, reservation.Currency, 1)
			}
			releaseKey := billingReservationReleaseIdempotencyKey(reserveKey)
			locked, err := lockWalletAccount(ctx, tx, reservation.AccountID)
			if err != nil {
				if err == pgx.ErrNoRows {
					continue
				}
				return err
			}
			var alreadyReleased bool
			if err := tx.QueryRow(ctx, `
SELECT EXISTS (
  SELECT 1
  FROM gateway_wallet_transactions
  WHERE account_id = $1::uuid
    AND idempotency_key = $2
)`, reservation.AccountID, releaseKey).Scan(&alreadyReleased); err != nil {
				return err
			}
			if alreadyReleased {
				continue
			}
			var storedReservedAmount string
			if err := tx.QueryRow(ctx, `
SELECT COALESCE((
  SELECT amount::text
  FROM gateway_wallet_transactions
  WHERE account_id = $1::uuid
    AND idempotency_key = $2
    AND transaction_type = 'reserve'
  LIMIT 1
), '0')`, reservation.AccountID, reserveKey).Scan(&storedReservedAmount); err != nil {
				return err
			}
			var positive bool
			if err := tx.QueryRow(ctx, `SELECT $1::numeric > 0`, storedReservedAmount).Scan(&positive); err != nil {
				return err
			}
			if !positive {
				continue
			}

			var balanceBefore string
			var frozenBefore string
			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 balance::text, (frozen_balance + $2::numeric)::text, frozen_balance::text`,
				locked.ID, storedReservedAmount).Scan(&balanceBefore, &frozenBefore, &frozenAfter); err != nil {
				return err
			}
			metadata, _ := json.Marshal(map[string]any{
				"taskId":       reservation.TaskID,
				"reason":       reason,
				"reserved":     storedReservedAmount,
				"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`,
				locked.ID,
				locked.GatewayTenantID,
				locked.GatewayUserID,
				storedReservedAmount,
				balanceBefore,
				releaseKey,
				reservation.TaskID,
				string(metadata),
			); err != nil {
				return err
			}
		}
		for taskID := range taskIDs {
			if _, err := tx.Exec(ctx, `
UPDATE gateway_tasks task
SET reservation_amount = COALESCE((
      SELECT SUM(reserve.amount)
      FROM gateway_wallet_transactions reserve
      WHERE reserve.reference_type = 'gateway_task'
        AND reserve.reference_id = task.id::text
        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'
        )
    ), 0),
    billing_status = CASE
      WHEN status IN ('failed', 'cancelled') THEN 'released'
      WHEN status = 'succeeded' THEN billing_status
      ELSE 'not_started'
    END,
    billing_updated_at = now(),
    updated_at = now()
WHERE id = $1::uuid`, taskID); err != nil {
				return err
			}
		}
		return nil
	})
}

func (s *Store) GetWalletSummary(ctx context.Context, user *auth.User, currency string) (WalletSummary, error) {
	gatewayUserID := localGatewayUserID(user)
	if gatewayUserID == "" {
		account := GatewayWalletAccount{
			Currency: normalizeWalletCurrency(currency),
			Status:   "active",
		}
		return WalletSummary{Accounts: []GatewayWalletAccount{account}, PrimaryAccount: account}, nil
	}
	primary, err := s.ensureWalletAccount(ctx, s.pool, gatewayUserID, currency)
	if err != nil {
		return WalletSummary{}, err
	}
	rows, err := s.pool.Query(ctx, `
SELECT id::text, COALESCE(gateway_tenant_id::text, ''), gateway_user_id::text,
       COALESCE(tenant_id, ''), COALESCE(tenant_key, ''), COALESCE(user_id, ''),
       currency, balance::float8, frozen_balance::float8, total_recharged::float8,
       total_spent::float8, status, metadata, created_at, updated_at
FROM gateway_wallet_accounts
WHERE gateway_user_id = $1::uuid
ORDER BY CASE WHEN currency = $2 THEN 0 WHEN currency = 'resource' THEN 1 ELSE 2 END, currency ASC`, gatewayUserID, primary.Currency)
	if err != nil {
		return WalletSummary{}, err
	}
	defer rows.Close()

	accounts := make([]GatewayWalletAccount, 0)
	for rows.Next() {
		account, err := scanWalletAccount(rows)
		if err != nil {
			return WalletSummary{}, err
		}
		accounts = append(accounts, account)
	}
	if err := rows.Err(); err != nil {
		return WalletSummary{}, err
	}
	if len(accounts) == 0 {
		accounts = append(accounts, primary)
	}
	return WalletSummary{Accounts: accounts, PrimaryAccount: accounts[0]}, nil
}

func (s *Store) ListWalletTransactions(ctx context.Context, user *auth.User, filter WalletTransactionListFilter) (WalletTransactionListResult, error) {
	page := filter.Page
	if page <= 0 {
		page = 1
	}
	pageSize := filter.PageSize
	if pageSize <= 0 {
		pageSize = 50
	}
	if pageSize > 100 {
		pageSize = 100
	}
	gatewayUserID := localGatewayUserID(user)
	if gatewayUserID == "" {
		return WalletTransactionListResult{Items: []GatewayWalletTransaction{}, Page: page, PageSize: pageSize}, nil
	}
	queryPattern := ""
	if query := strings.TrimSpace(filter.Query); query != "" {
		queryPattern = "%" + query + "%"
	}
	args := []any{
		gatewayUserID,
		queryPattern,
		strings.TrimSpace(filter.Direction),
		strings.TrimSpace(filter.TransactionType),
		nullableTaskListTime(filter.CreatedFrom),
		nullableTaskListTime(filter.CreatedTo),
	}
	whereSQL := `
WHERE a.gateway_user_id = $1::uuid
  AND (
    NULLIF($2, '') IS NULL
    OR t.id::text ILIKE $2
    OR COALESCE(t.reference_id, '') ILIKE $2
    OR COALESCE(t.reference_type, '') ILIKE $2
    OR COALESCE(t.idempotency_key, '') ILIKE $2
    OR t.transaction_type ILIKE $2
    OR t.direction ILIKE $2
    OR COALESCE(task.id::text, '') ILIKE $2
    OR COALESCE(task.request_id, '') ILIKE $2
    OR COALESCE(task.kind, '') ILIKE $2
    OR COALESCE(task.model, '') ILIKE $2
	    OR COALESCE(task.requested_model, '') ILIKE $2
	    OR COALESCE(task.resolved_model, '') ILIKE $2
	    OR COALESCE(task.model_type, '') ILIKE $2
	    OR COALESCE(task.api_key_id, '') ILIKE $2
	    OR COALESCE(task.api_key_name, '') ILIKE $2
	    OR COALESCE(task.api_key_prefix, '') ILIKE $2
	    OR COALESCE(task.status, '') ILIKE $2
	    OR COALESCE(task.billing_summary->>'currency', '') ILIKE $2
	    OR COALESCE(task.billing_summary->>'totalAmount', '') ILIKE $2
	    OR COALESCE(attempt.client_id, '') ILIKE $2
	    OR COALESCE(attempt.request_id, '') ILIKE $2
	    OR COALESCE(platform.provider, '') ILIKE $2
	    OR COALESCE(platform.platform_key, '') ILIKE $2
	    OR COALESCE(platform.name, '') ILIKE $2
    OR COALESCE(platform_model.model_name, '') ILIKE $2
    OR COALESCE(platform_model.provider_model_name, '') ILIKE $2
    OR COALESCE(platform_model.model_alias, '') ILIKE $2
    OR COALESCE(platform_model.display_name, '') ILIKE $2
    OR COALESCE(task.metrics->>'provider', '') ILIKE $2
    OR COALESCE(task.metrics->>'platformName', '') ILIKE $2
    OR COALESCE(task.metrics->>'modelAlias', '') ILIKE $2
    OR COALESCE(task.metrics->>'providerModel', '') ILIKE $2
  )
  AND (NULLIF($3, '') IS NULL OR t.direction = $3)
  AND (NULLIF($4, '') IS NULL OR t.transaction_type = $4)
  AND ($5::timestamptz IS NULL OR t.created_at >= $5::timestamptz)
  AND ($6::timestamptz IS NULL OR t.created_at <= $6::timestamptz)`
	var total int
	if err := s.pool.QueryRow(ctx, `
SELECT count(*)
	FROM gateway_wallet_transactions t
	JOIN gateway_wallet_accounts a ON a.id = t.account_id
	LEFT JOIN gateway_tasks task ON t.reference_type = 'gateway_task' AND t.reference_id = task.id::text
	LEFT JOIN LATERAL (
	  SELECT platform_id, platform_model_id, client_id, request_id
	  FROM gateway_task_attempts
	  WHERE task_id = task.id
	  ORDER BY attempt_no DESC, started_at DESC
	  LIMIT 1
) attempt ON true
LEFT JOIN integration_platforms platform ON platform.id = attempt.platform_id
LEFT JOIN platform_models platform_model ON platform_model.id = attempt.platform_model_id
`+whereSQL, args...).Scan(&total); err != nil {
		return WalletTransactionListResult{}, err
	}
	offset := (page - 1) * pageSize
	queryArgs := append(args, pageSize, offset)
	rows, err := s.pool.Query(ctx, `
SELECT t.id::text, t.account_id::text, a.currency, COALESCE(t.gateway_tenant_id::text, ''),
       COALESCE(t.gateway_user_id::text, ''), t.direction, t.transaction_type,
       t.amount::float8, t.balance_before::float8, t.balance_after::float8,
       COALESCE(t.idempotency_key, ''), COALESCE(t.reference_type, ''),
       COALESCE(t.reference_id, ''),
       t.metadata || jsonb_strip_nulls(jsonb_build_object(
         'taskId', task.id::text,
         'kind', task.kind,
         'model', COALESCE(NULLIF(task.requested_model, ''), NULLIF(task.model, ''), NULLIF(platform_model.model_alias, ''), NULLIF(task.metrics->>'modelAlias', ''), NULLIF(task.resolved_model, ''), NULLIF(platform_model.model_name, ''), NULLIF(task.metrics->>'modelName', '')),
         'requestedModel', COALESCE(NULLIF(task.requested_model, ''), NULLIF(task.model, '')),
         'resolvedModel', COALESCE(NULLIF(platform_model.model_alias, ''), NULLIF(task.metrics->>'modelAlias', ''), NULLIF(task.resolved_model, ''), NULLIF(platform_model.model_name, ''), NULLIF(task.metrics->>'modelName', '')),
         'modelName', COALESCE(NULLIF(platform_model.model_name, ''), NULLIF(task.metrics->>'modelName', ''), NULLIF(task.resolved_model, '')),
         'modelAlias', COALESCE(NULLIF(platform_model.model_alias, ''), NULLIF(task.metrics->>'modelAlias', '')),
         'modelType', task.model_type,
         'taskStatus', task.status,
         'runMode', task.run_mode,
         'requestId', COALESCE(task.request_id, attempt.request_id),
         'apiKeyId', task.api_key_id,
         'apiKeyName', task.api_key_name,
         'apiKeyPrefix', task.api_key_prefix,
         'provider', COALESCE(platform.provider, task.metrics->>'provider'),
         'platformId', COALESCE(platform.id::text, task.metrics->>'platformId'),
         'platformName', COALESCE(platform.name, task.metrics->>'platformName'),
         'platformKey', platform.platform_key,
         'platformModelId', COALESCE(platform_model.id::text, task.metrics->>'platformModelId'),
         'platformModelName', platform_model.model_name,
         'platformModelAlias', platform_model.model_alias,
         'providerModel', COALESCE(platform_model.provider_model_name, task.metrics->>'providerModel'),
         'clientId', attempt.client_id,
         'usage', CASE WHEN task.id IS NULL THEN NULL ELSE COALESCE(task.usage, attempt.usage, '{}'::jsonb) END,
         'billings', CASE WHEN task.id IS NULL THEN NULL ELSE COALESCE(task.billings, '[]'::jsonb) END,
         'billingSummary', CASE WHEN task.id IS NULL THEN NULL ELSE COALESCE(task.billing_summary, '{}'::jsonb) END,
         'finalChargeAmount', CASE WHEN task.id IS NULL THEN NULL ELSE COALESCE(task.final_charge_amount, 0)::float8 END,
         'responseStartedAt', COALESCE(task.response_started_at::text, attempt.response_started_at::text),
         'responseFinishedAt', COALESCE(task.response_finished_at::text, attempt.response_finished_at::text),
         'responseDurationMs', COALESCE(task.response_duration_ms, attempt.response_duration_ms)
       )), t.created_at
	FROM gateway_wallet_transactions t
	JOIN gateway_wallet_accounts a ON a.id = t.account_id
	LEFT JOIN gateway_tasks task ON t.reference_type = 'gateway_task' AND t.reference_id = task.id::text
	LEFT JOIN LATERAL (
	  SELECT platform_id, platform_model_id, client_id, request_id, usage, response_started_at,
	         response_finished_at, response_duration_ms
	  FROM gateway_task_attempts
	  WHERE task_id = task.id
	  ORDER BY attempt_no DESC, started_at DESC
	  LIMIT 1
) attempt ON true
LEFT JOIN integration_platforms platform ON platform.id = attempt.platform_id
LEFT JOIN platform_models platform_model ON platform_model.id = attempt.platform_model_id
`+whereSQL+`
ORDER BY t.created_at DESC, t.id DESC
LIMIT $7 OFFSET $8`, queryArgs...)
	if err != nil {
		return WalletTransactionListResult{}, err
	}
	defer rows.Close()

	items := make([]GatewayWalletTransaction, 0)
	for rows.Next() {
		item, err := scanWalletTransactionWithCurrency(rows)
		if err != nil {
			return WalletTransactionListResult{}, err
		}
		items = append(items, item)
	}
	if err := rows.Err(); err != nil {
		return WalletTransactionListResult{}, err
	}
	return WalletTransactionListResult{Items: items, Total: total, Page: page, PageSize: pageSize}, nil
}

func (s *Store) SetUserWalletBalance(ctx context.Context, input WalletBalanceAdjustmentInput) (WalletAdjustmentResult, error) {
	var result WalletAdjustmentResult
	err := s.InTx(ctx, func(tx Tx) error {
		next, err := s.SetUserWalletBalanceTx(ctx, tx, input)
		if err != nil {
			return err
		}
		result = next
		return nil
	})
	return result, err
}

func (s *Store) SetUserWalletBalanceTx(ctx context.Context, tx Tx, input WalletBalanceAdjustmentInput) (WalletAdjustmentResult, error) {
	input.GatewayUserID = strings.TrimSpace(input.GatewayUserID)
	if input.GatewayUserID == "" {
		return WalletAdjustmentResult{}, ErrLocalUserRequired
	}
	targetInput := strings.TrimSpace(input.BalanceText)
	if targetInput == "" {
		targetInput = strconv.FormatFloat(input.Balance, 'f', 9, 64)
	}
	targetBalance, err := canonicalWalletAmount(ctx, tx, targetInput)
	if err != nil {
		return WalletAdjustmentResult{}, err
	}
	var nonnegative bool
	if err := tx.QueryRow(ctx, `SELECT $1::numeric >= 0`, targetBalance).Scan(&nonnegative); err != nil {
		return WalletAdjustmentResult{}, err
	}
	if !nonnegative {
		return WalletAdjustmentResult{}, ErrInvalidWalletAmount
	}
	account, err := s.ensureWalletAccount(ctx, tx, input.GatewayUserID, input.Currency)
	if err != nil {
		return WalletAdjustmentResult{}, err
	}
	var locked GatewayWalletAccount
	locked, err = scanWalletAccount(tx.QueryRow(ctx, `
SELECT id::text, COALESCE(gateway_tenant_id::text, ''), gateway_user_id::text,
       COALESCE(tenant_id, ''), COALESCE(tenant_key, ''), COALESCE(user_id, ''),
       currency, balance::float8, frozen_balance::float8, total_recharged::float8,
       total_spent::float8, status, metadata, created_at, updated_at
FROM gateway_wallet_accounts
WHERE id = $1::uuid
FOR UPDATE`, account.ID))
	if err != nil {
		return WalletAdjustmentResult{}, err
	}
	before := locked
	var balanceBefore string
	var frozenBalance string
	if err := tx.QueryRow(ctx, `
SELECT balance::text, frozen_balance::text
FROM gateway_wallet_accounts
WHERE id = $1::uuid`, locked.ID).Scan(&balanceBefore, &frozenBalance); err != nil {
		return WalletAdjustmentResult{}, err
	}
	var unchanged bool
	var belowFrozen bool
	if err := tx.QueryRow(ctx, `
SELECT $1::numeric = $2::numeric, $1::numeric < $3::numeric`,
		targetBalance, balanceBefore, frozenBalance).Scan(&unchanged, &belowFrozen); err != nil {
		return WalletAdjustmentResult{}, err
	}
	if unchanged {
		return WalletAdjustmentResult{}, ErrWalletBalanceUnchanged
	}
	if belowFrozen {
		return WalletAdjustmentResult{}, ErrBalanceBelowFrozen
	}
	direction := "credit"
	if strings.HasPrefix(balanceBefore, "-") {
		return WalletAdjustmentResult{}, ErrInvalidWalletAmount
	}
	var targetGreater bool
	if err := tx.QueryRow(ctx, `SELECT $1::numeric > $2::numeric`, targetBalance, balanceBefore).Scan(&targetGreater); err != nil {
		return WalletAdjustmentResult{}, err
	}
	if !targetGreater {
		direction = "debit"
	}
	var amount string
	if err := tx.QueryRow(ctx, `SELECT abs($1::numeric - $2::numeric)::text`, targetBalance, balanceBefore).Scan(&amount); err != nil {
		return WalletAdjustmentResult{}, err
	}
	reason := strings.TrimSpace(input.Reason)
	if reason == "" {
		reason = "后台余额调整"
	}
	tag, err := tx.Exec(ctx, `
UPDATE gateway_wallet_accounts
SET balance = $2::numeric,
    total_recharged = total_recharged + CASE WHEN $3 = 'credit' THEN $4::numeric ELSE 0 END,
    updated_at = now()
WHERE id = $1::uuid
  AND $2::numeric >= frozen_balance`,
		locked.ID,
		targetBalance,
		direction,
		amount,
	)
	if err != nil {
		return WalletAdjustmentResult{}, err
	}
	if tag.RowsAffected() != 1 {
		return WalletAdjustmentResult{}, ErrBalanceBelowFrozen
	}
	metadata := mergeObjects(input.Metadata, map[string]any{
		"reason":          reason,
		"previousBalance": balanceBefore,
		"targetBalance":   targetBalance,
	})
	metadataJSON, _ := json.Marshal(emptyObjectIfNil(metadata))
	transaction, err := scanWalletTransaction(tx.QueryRow(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, $4, 'admin_adjust',
  $5::numeric, $6::numeric, $7::numeric, NULLIF($8, ''), 'gateway_user', $9, $10::jsonb
)
RETURNING id::text, account_id::text, COALESCE(gateway_tenant_id::text, ''), COALESCE(gateway_user_id::text, ''),
          direction, transaction_type, amount::float8, balance_before::float8, balance_after::float8,
          COALESCE(idempotency_key, ''), COALESCE(reference_type, ''), COALESCE(reference_id, ''),
          metadata, created_at`,
		locked.ID,
		locked.GatewayTenantID,
		locked.GatewayUserID,
		direction,
		amount,
		balanceBefore,
		targetBalance,
		strings.TrimSpace(input.IdempotencyKey),
		locked.GatewayUserID,
		string(metadataJSON),
	))
	if err != nil {
		return WalletAdjustmentResult{}, err
	}
	locked, err = s.ensureWalletAccount(ctx, tx, input.GatewayUserID, input.Currency)
	if err != nil {
		return WalletAdjustmentResult{}, err
	}
	return WalletAdjustmentResult{Account: locked, Before: before, Transaction: transaction}, nil
}

func (s *Store) RechargeUserWalletBalance(ctx context.Context, input WalletRechargeInput) (WalletAdjustmentResult, error) {
	var result WalletAdjustmentResult
	err := s.InTx(ctx, func(tx Tx) error {
		next, err := s.RechargeUserWalletBalanceTx(ctx, tx, input)
		if err != nil {
			return err
		}
		result = next
		return nil
	})
	return result, err
}

func (s *Store) RechargeUserWalletBalanceTx(ctx context.Context, tx Tx, input WalletRechargeInput) (WalletAdjustmentResult, error) {
	input.GatewayUserID = strings.TrimSpace(input.GatewayUserID)
	if input.GatewayUserID == "" {
		return WalletAdjustmentResult{}, ErrLocalUserRequired
	}
	amountInput := strings.TrimSpace(input.AmountText)
	if amountInput == "" {
		amountInput = strconv.FormatFloat(input.Amount, 'f', 9, 64)
	}
	amount, err := canonicalWalletAmount(ctx, tx, amountInput)
	if err != nil {
		return WalletAdjustmentResult{}, err
	}
	var positive bool
	if err := tx.QueryRow(ctx, `SELECT $1::numeric > 0`, amount).Scan(&positive); err != nil {
		return WalletAdjustmentResult{}, err
	}
	if !positive {
		return WalletAdjustmentResult{}, ErrInvalidWalletAmount
	}
	account, err := s.ensureWalletAccount(ctx, tx, input.GatewayUserID, input.Currency)
	if err != nil {
		return WalletAdjustmentResult{}, err
	}
	locked, err := scanWalletAccount(tx.QueryRow(ctx, `
SELECT id::text, COALESCE(gateway_tenant_id::text, ''), gateway_user_id::text,
       COALESCE(tenant_id, ''), COALESCE(tenant_key, ''), COALESCE(user_id, ''),
       currency, balance::float8, frozen_balance::float8, total_recharged::float8,
       total_spent::float8, status, metadata, created_at, updated_at
FROM gateway_wallet_accounts
WHERE id = $1::uuid
FOR UPDATE`, account.ID))
	if err != nil {
		return WalletAdjustmentResult{}, err
	}
	before := locked
	var balanceBefore string
	if err := tx.QueryRow(ctx, `SELECT balance::text FROM gateway_wallet_accounts WHERE id = $1::uuid`, locked.ID).Scan(&balanceBefore); err != nil {
		return WalletAdjustmentResult{}, err
	}
	reason := strings.TrimSpace(input.Reason)
	if reason == "" {
		reason = "后台余额充值"
	}
	var nextBalance string
	if err := tx.QueryRow(ctx, `
UPDATE gateway_wallet_accounts
SET balance = balance + $2::numeric,
    total_recharged = total_recharged + $2::numeric,
    updated_at = now()
WHERE id = $1::uuid
RETURNING balance::text`,
		locked.ID,
		amount,
	).Scan(&nextBalance); err != nil {
		return WalletAdjustmentResult{}, err
	}
	metadata := mergeObjects(input.Metadata, map[string]any{
		"reason":          reason,
		"previousBalance": balanceBefore,
		"rechargeAmount":  amount,
		"targetBalance":   nextBalance,
	})
	metadataJSON, _ := json.Marshal(emptyObjectIfNil(metadata))
	transaction, err := scanWalletTransaction(tx.QueryRow(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', 'recharge',
  $4::numeric, $5::numeric, $6::numeric, NULLIF($7, ''), 'gateway_user', $8, $9::jsonb
)
RETURNING id::text, account_id::text, COALESCE(gateway_tenant_id::text, ''), COALESCE(gateway_user_id::text, ''),
          direction, transaction_type, amount::float8, balance_before::float8, balance_after::float8,
          COALESCE(idempotency_key, ''), COALESCE(reference_type, ''), COALESCE(reference_id, ''),
          metadata, created_at`,
		locked.ID,
		locked.GatewayTenantID,
		locked.GatewayUserID,
		amount,
		balanceBefore,
		nextBalance,
		strings.TrimSpace(input.IdempotencyKey),
		locked.GatewayUserID,
		string(metadataJSON),
	))
	if err != nil {
		return WalletAdjustmentResult{}, err
	}
	locked, err = s.ensureWalletAccount(ctx, tx, input.GatewayUserID, input.Currency)
	if err != nil {
		return WalletAdjustmentResult{}, err
	}
	return WalletAdjustmentResult{Account: locked, Before: before, Transaction: transaction}, nil
}

func canonicalWalletAmount(ctx context.Context, q Tx, value string) (string, error) {
	var canonical string
	if err := q.QueryRow(ctx, `SELECT $1::numeric(38, 9)::text`, strings.TrimSpace(value)).Scan(&canonical); err != nil {
		return "", fmt.Errorf("%w: %v", ErrInvalidWalletAmount, err)
	}
	return canonical, nil
}

func (s *Store) ensureWalletAccount(ctx context.Context, q Tx, gatewayUserID string, currency string) (GatewayWalletAccount, error) {
	currency = normalizeWalletCurrency(currency)
	if _, err := q.Exec(ctx, `
INSERT INTO gateway_wallet_accounts (
  gateway_tenant_id, gateway_user_id, tenant_id, tenant_key, user_id, currency
)
SELECT gateway_tenant_id, id, NULLIF(tenant_id, ''), NULLIF(tenant_key, ''),
       COALESCE(NULLIF(external_user_id, ''), id::text), $2
FROM gateway_users
WHERE id = $1::uuid
  AND deleted_at IS NULL
ON CONFLICT (gateway_user_id, currency) DO UPDATE
SET gateway_tenant_id = COALESCE(gateway_wallet_accounts.gateway_tenant_id, EXCLUDED.gateway_tenant_id),
    tenant_id = COALESCE(NULLIF(gateway_wallet_accounts.tenant_id, ''), EXCLUDED.tenant_id),
    tenant_key = COALESCE(NULLIF(gateway_wallet_accounts.tenant_key, ''), EXCLUDED.tenant_key),
    user_id = COALESCE(NULLIF(gateway_wallet_accounts.user_id, ''), EXCLUDED.user_id),
    updated_at = now()
WHERE gateway_wallet_accounts.gateway_tenant_id IS NULL
   OR COALESCE(gateway_wallet_accounts.tenant_id, '') = ''
   OR COALESCE(gateway_wallet_accounts.tenant_key, '') = ''
   OR COALESCE(gateway_wallet_accounts.user_id, '') = ''`, gatewayUserID, currency); err != nil {
		return GatewayWalletAccount{}, err
	}
	if err := ensureWalletAccountAuditGuard(ctx, q, gatewayUserID, currency); err != nil {
		return GatewayWalletAccount{}, err
	}
	account, err := scanWalletAccount(q.QueryRow(ctx, `
SELECT id::text, COALESCE(gateway_tenant_id::text, ''), gateway_user_id::text,
       COALESCE(tenant_id, ''), COALESCE(tenant_key, ''), COALESCE(user_id, ''),
       currency, balance::float8, frozen_balance::float8, total_recharged::float8,
       total_spent::float8, status, metadata, created_at, updated_at
FROM gateway_wallet_accounts
WHERE gateway_user_id = $1::uuid
  AND currency = $2`, gatewayUserID, currency))
	if err != nil {
		if err == pgx.ErrNoRows {
			return GatewayWalletAccount{}, pgx.ErrNoRows
		}
		return GatewayWalletAccount{}, err
	}
	return account, nil
}

func ensureWalletAccountAuditGuard(ctx context.Context, q Tx, gatewayUserID string, currency string) error {
	_, err := q.Exec(ctx, `
INSERT INTO gateway_wallet_account_audit_guards (account_id)
SELECT id
FROM gateway_wallet_accounts
WHERE gateway_user_id = $1::uuid
  AND currency = $2
ON CONFLICT (account_id) DO NOTHING`, gatewayUserID, normalizeWalletCurrency(currency))
	return err
}

func lockWalletAccount(ctx context.Context, tx pgx.Tx, accountID string) (GatewayWalletAccount, error) {
	return scanWalletAccount(tx.QueryRow(ctx, `
SELECT id::text, COALESCE(gateway_tenant_id::text, ''), gateway_user_id::text,
       COALESCE(tenant_id, ''), COALESCE(tenant_key, ''), COALESCE(user_id, ''),
       currency, balance::float8, frozen_balance::float8, total_recharged::float8,
       total_spent::float8, status, metadata, created_at, updated_at
FROM gateway_wallet_accounts
WHERE id = $1::uuid
FOR UPDATE`, accountID))
}

func activeWalletReservation(ctx context.Context, tx pgx.Tx, accountID string, taskID string) (string, float64, error) {
	var key string
	var amount float64
	err := tx.QueryRow(ctx, `
SELECT COALESCE(t.idempotency_key, ''), t.amount::float8
FROM gateway_wallet_transactions t
WHERE t.account_id = $1::uuid
  AND t.reference_type = 'gateway_task'
  AND t.reference_id = $2
  AND t.transaction_type = 'reserve'
  AND COALESCE(t.idempotency_key, '') <> ''
  AND NOT EXISTS (
    SELECT 1
    FROM gateway_wallet_transactions r
    WHERE r.account_id = t.account_id
      AND r.transaction_type = 'release'
      AND r.idempotency_key = t.idempotency_key || ':release'
  )
ORDER BY t.created_at DESC
LIMIT 1`, accountID, taskID).Scan(&key, &amount)
	if err == pgx.ErrNoRows {
		return "", 0, nil
	}
	if err != nil {
		return "", 0, err
	}
	return key, roundMoney(amount), nil
}

func activeWalletReservationExact(ctx context.Context, tx pgx.Tx, accountID string, taskID string) (string, string, error) {
	var key string
	var amount string
	err := tx.QueryRow(ctx, `
SELECT COALESCE(t.idempotency_key, ''), t.amount::text
FROM gateway_wallet_transactions t
WHERE t.account_id = $1::uuid
  AND t.reference_type = 'gateway_task'
  AND t.reference_id = $2
  AND t.transaction_type = 'reserve'
  AND COALESCE(t.idempotency_key, '') <> ''
  AND NOT EXISTS (
    SELECT 1 FROM gateway_wallet_transactions release
    WHERE release.account_id = t.account_id
      AND release.transaction_type = 'release'
      AND release.idempotency_key = t.idempotency_key || ':release'
  )
ORDER BY t.created_at DESC
LIMIT 1`, accountID, taskID).Scan(&key, &amount)
	if errors.Is(err, pgx.ErrNoRows) {
		return "", "", nil
	}
	return key, amount, err
}

func nextWalletReservationSequence(ctx context.Context, tx pgx.Tx, accountID string, taskID string) (int, error) {
	var count int
	if err := tx.QueryRow(ctx, `
SELECT COUNT(*)::int
FROM gateway_wallet_transactions
WHERE account_id = $1::uuid
  AND reference_type = 'gateway_task'
  AND reference_id = $2
  AND transaction_type = 'reserve'`, accountID, taskID).Scan(&count); err != nil {
		return 0, err
	}
	return count + 1, nil
}

func walletBillingAmounts(billings []any) map[string]float64 {
	amounts := map[string]float64{}
	for _, raw := range billings {
		line, _ := raw.(map[string]any)
		if line == nil {
			continue
		}
		amount := roundMoney(walletFloat(line["amount"]))
		if amount <= 0 {
			continue
		}
		currency := normalizeWalletCurrency(walletString(line["currency"]))
		amounts[currency] = roundMoney(amounts[currency] + amount)
	}
	return amounts
}

func taskGatewayUserID(task GatewayTask, user *auth.User) string {
	return firstNonEmpty(strings.TrimSpace(task.GatewayUserID), localGatewayUserID(user))
}

func billingReservationIdempotencyKey(taskID string, currency string, sequence int) string {
	if sequence <= 0 {
		sequence = 1
	}
	return "task:" + strings.TrimSpace(taskID) + ":wallet-reservation:" + normalizeWalletCurrency(currency) + ":" + strconv.Itoa(sequence)
}

func billingReservationReleaseIdempotencyKey(reservationKey string) string {
	return strings.TrimSpace(reservationKey) + ":release"
}

func walletString(value any) string {
	if text, ok := value.(string); ok {
		return strings.TrimSpace(text)
	}
	return ""
}

func walletFloat(value any) float64 {
	switch typed := value.(type) {
	case float64:
		return typed
	case float32:
		return float64(typed)
	case int:
		return float64(typed)
	case int64:
		return float64(typed)
	case json.Number:
		next, _ := typed.Float64()
		return next
	case string:
		next := strings.TrimSpace(typed)
		if next == "" {
			return 0
		}
		parsed, _ := strconv.ParseFloat(next, 64)
		return parsed
	default:
		return 0
	}
}

type billingModelIdentity struct {
	Model             string
	RequestedModel    string
	ResolvedModel     string
	ModelName         string
	ModelAlias        string
	ProviderModelName string
}

func taskBillingModelIdentity(task GatewayTask) billingModelIdentity {
	modelAlias := firstNonEmpty(
		taskBillingString(task.Metrics["modelAlias"]),
		firstBillingLineString(task.Billings, "modelAlias"),
	)
	modelName := firstNonEmpty(
		taskBillingString(task.Metrics["modelName"]),
		taskBillingString(task.Metrics["resolvedModel"]),
		task.ResolvedModel,
		firstBillingLineString(task.Billings, "model"),
	)
	providerModelName := firstNonEmpty(
		taskBillingString(task.Metrics["providerModel"]),
		firstBillingLineString(task.Billings, "providerModel"),
	)
	requestedModel := firstNonEmpty(task.RequestedModel, task.Model)
	systemModel := firstNonEmpty(modelAlias, modelName, task.ResolvedModel, requestedModel)
	return billingModelIdentity{
		Model:             firstNonEmpty(requestedModel, systemModel),
		RequestedModel:    requestedModel,
		ResolvedModel:     systemModel,
		ModelName:         modelName,
		ModelAlias:        modelAlias,
		ProviderModelName: providerModelName,
	}
}

func firstBillingLineString(billings []any, key string) string {
	for _, raw := range billings {
		line, _ := raw.(map[string]any)
		if line == nil {
			continue
		}
		if value := strings.TrimSpace(taskBillingString(line[key])); value != "" {
			return value
		}
	}
	return ""
}

func normalizeWalletCurrency(currency string) string {
	currency = strings.TrimSpace(currency)
	if currency == "" {
		return "resource"
	}
	return currency
}

func scanWalletAccount(row scanner) (GatewayWalletAccount, error) {
	var item GatewayWalletAccount
	var metadata []byte
	if err := row.Scan(
		&item.ID,
		&item.GatewayTenantID,
		&item.GatewayUserID,
		&item.TenantID,
		&item.TenantKey,
		&item.UserID,
		&item.Currency,
		&item.Balance,
		&item.FrozenBalance,
		&item.TotalRecharged,
		&item.TotalSpent,
		&item.Status,
		&metadata,
		&item.CreatedAt,
		&item.UpdatedAt,
	); err != nil {
		return GatewayWalletAccount{}, err
	}
	item.Metadata = decodeObject(metadata)
	return item, nil
}

func scanWalletTransaction(row scanner) (GatewayWalletTransaction, error) {
	var item GatewayWalletTransaction
	var metadata []byte
	if err := row.Scan(
		&item.ID,
		&item.AccountID,
		&item.GatewayTenantID,
		&item.GatewayUserID,
		&item.Direction,
		&item.TransactionType,
		&item.Amount,
		&item.BalanceBefore,
		&item.BalanceAfter,
		&item.IdempotencyKey,
		&item.ReferenceType,
		&item.ReferenceID,
		&metadata,
		&item.CreatedAt,
	); err != nil {
		return GatewayWalletTransaction{}, err
	}
	item.Metadata = decodeObject(metadata)
	return item, nil
}

func scanWalletTransactionWithCurrency(row scanner) (GatewayWalletTransaction, error) {
	var item GatewayWalletTransaction
	var metadata []byte
	if err := row.Scan(
		&item.ID,
		&item.AccountID,
		&item.Currency,
		&item.GatewayTenantID,
		&item.GatewayUserID,
		&item.Direction,
		&item.TransactionType,
		&item.Amount,
		&item.BalanceBefore,
		&item.BalanceAfter,
		&item.IdempotencyKey,
		&item.ReferenceType,
		&item.ReferenceID,
		&metadata,
		&item.CreatedAt,
	); err != nil {
		return GatewayWalletTransaction{}, err
	}
	item.Metadata = decodeObject(metadata)
	return item, nil
}

func decodeWalletAccounts(data []byte) []GatewayWalletAccount {
	if len(data) == 0 {
		return nil
	}
	var items []GatewayWalletAccount
	if err := json.Unmarshal(data, &items); err != nil {
		return nil
	}
	return items
}
