path: root/internal/billing/operations.go

package billing

import (
	"context"
	"encoding/csv"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"strconv"
	"strings"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/stripe/stripe-go/v86"
)

func (s *Service) CreatePortalSession(ctx context.Context, tenantID string) (PortalResult, error) {
	if !s.stripeEnabled || s.stripeClient == nil {
		return PortalResult{}, ErrStripeDisabled
	}
	var customerID string
	if err := s.db.QueryRow(ctx, `SELECT stripe_customer_id FROM stripe_customers WHERE tenant_id=$1`, tenantID).Scan(&customerID); err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return PortalResult{}, errors.New("no Stripe customer exists for this account")
		}
		return PortalResult{}, err
	}
	session, err := s.stripeClient.V1BillingPortalSessions.Create(ctx, &stripe.BillingPortalSessionCreateParams{
		Customer: stripe.String(customerID), ReturnURL: stripe.String(s.stripePortalReturnURL),
	})
	if err != nil {
		return PortalResult{}, fmt.Errorf("create Stripe customer portal session: %w", err)
	}
	if session.URL == "" {
		return PortalResult{}, errors.New("Stripe returned an incomplete portal session")
	}
	return PortalResult{URL: session.URL}, nil
}

func (s *Service) RetryCheckout(ctx context.Context, tenantID, orderID, email string) (CheckoutResult, error) {
	var amount int64
	var status string
	if err := s.db.QueryRow(ctx, `SELECT amount_minor,status FROM topup_orders WHERE id=$1 AND tenant_id=$2`, orderID, tenantID).Scan(&amount, &status); err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return CheckoutResult{}, ErrTopUpOrderNotFound
		}
		return CheckoutResult{}, err
	}
	if status != "failed" && status != "expired" {
		return CheckoutResult{}, errors.New("only failed or expired top-ups can be retried")
	}
	return s.CreateCheckout(ctx, CheckoutInput{TenantID: tenantID, AmountMinor: amount, CustomerEmail: email})
}

// ResolveMissingTopUp closes an uncredited local order only after reconciliation
// proved that its Checkout Session does not exist in the configured Stripe account.
func (s *Service) ResolveMissingTopUp(ctx context.Context, tenantID, orderID string, input ResolveMissingTopUpInput, actor ResolutionActor) (TopUpOrder, error) {
	reason := normalizeDescription(input.Reason)
	if reason == "" {
		return TopUpOrder{}, fmt.Errorf("%w: a resolution reason is required", ErrCannotResolveTopUp)
	}
	tx, err := s.db.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.ReadCommitted})
	if err != nil {
		return TopUpOrder{}, err
	}
	defer tx.Rollback(ctx)
	var status, reconciliationStatus, paymentIntentID string
	if err := tx.QueryRow(ctx, `SELECT status,reconciliation_status,COALESCE(stripe_payment_intent_id,'')
		FROM topup_orders WHERE id=$1 AND tenant_id=$2 FOR UPDATE`, orderID, tenantID).
		Scan(&status, &reconciliationStatus, &paymentIntentID); err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return TopUpOrder{}, ErrTopUpOrderNotFound
		}
		return TopUpOrder{}, err
	}
	if status != "pending" || reconciliationStatus != "missing" || paymentIntentID != "" {
		return TopUpOrder{}, fmt.Errorf("%w: only an uncredited pending order confirmed missing by reconciliation can be resolved", ErrCannotResolveTopUp)
	}
	var credits int64
	if err := tx.QueryRow(ctx, `SELECT count(*) FROM billing_ledger
		WHERE source_type='stripe_checkout' AND source_id=(SELECT stripe_session_id FROM topup_orders WHERE id=$1)`, orderID).Scan(&credits); err != nil {
		return TopUpOrder{}, err
	}
	if credits != 0 {
		return TopUpOrder{}, fmt.Errorf("%w: a credited top-up cannot be resolved as missing", ErrCannotResolveTopUp)
	}
	if actor.Type != "console_user" && actor.Type != "bootstrap" && actor.Type != "maintenance" {
		return TopUpOrder{}, fmt.Errorf("%w: a valid resolution actor is required", ErrCannotResolveTopUp)
	}
	if _, err := tx.Exec(ctx, `INSERT INTO billing_reconciliation_resolutions
		(topup_order_id,tenant_id,actor_id,actor_type,reason) VALUES ($1,$2,$3,$4,$5)`,
		orderID, tenantID, actor.ID, actor.Type, reason); err != nil {
		return TopUpOrder{}, err
	}
	if _, err := tx.Exec(ctx, `UPDATE topup_orders SET status='failed',reconciliation_status='resolved',
		reconciled_at=now(),reconciliation_error=$2 WHERE id=$1`, orderID, reason); err != nil {
		return TopUpOrder{}, err
	}
	if err := tx.Commit(ctx); err != nil {
		return TopUpOrder{}, err
	}
	return s.GetTopUpOrder(ctx, tenantID, orderID)
}

// ReverseMissingTopUpCredit preserves the original credit and adds an equal
// negative ledger entry when the configured Stripe account cannot prove the
// payment. It refuses to consume funds reserved for in-flight requests.
func (s *Service) ReverseMissingTopUpCredit(ctx context.Context, tenantID, orderID string, input ResolveMissingTopUpInput, actor ResolutionActor) (TopUpOrder, error) {
	reason := normalizeDescription(input.Reason)
	if reason == "" || (actor.Type != "console_user" && actor.Type != "bootstrap" && actor.Type != "maintenance") {
		return TopUpOrder{}, fmt.Errorf("%w: a reason and valid resolution actor are required", ErrCannotResolveTopUp)
	}
	tx, err := s.db.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.ReadCommitted})
	if err != nil {
		return TopUpOrder{}, err
	}
	defer tx.Rollback(ctx)
	var status, reconciliationStatus, paymentIntentID, currency, sessionID string
	var amount int64
	if err := tx.QueryRow(ctx, `SELECT status,reconciliation_status,COALESCE(stripe_payment_intent_id,''),
		currency,amount_micros,COALESCE(stripe_session_id,'') FROM topup_orders
		WHERE id=$1 AND tenant_id=$2 FOR UPDATE`, orderID, tenantID).
		Scan(&status, &reconciliationStatus, &paymentIntentID, &currency, &amount, &sessionID); err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return TopUpOrder{}, ErrTopUpOrderNotFound
		}
		return TopUpOrder{}, err
	}
	if status != "paid" || reconciliationStatus != "missing" || paymentIntentID != "" || sessionID == "" {
		return TopUpOrder{}, fmt.Errorf("%w: only a paid credit confirmed missing with no PaymentIntent can be reversed", ErrCannotResolveTopUp)
	}
	var originalCredits int64
	if err := tx.QueryRow(ctx, `SELECT COALESCE(sum(amount_micros),0) FROM billing_ledger
		WHERE tenant_id=$1 AND source_type='stripe_checkout' AND source_id=$2`, tenantID, sessionID).Scan(&originalCredits); err != nil {
		return TopUpOrder{}, err
	}
	if originalCredits != amount {
		return TopUpOrder{}, fmt.Errorf("%w: original Stripe credit does not match the order", ErrCannotResolveTopUp)
	}
	var balance, reserved int64
	if err := tx.QueryRow(ctx, `SELECT balance_micros,reserved_micros FROM tenant_wallets WHERE tenant_id=$1 FOR UPDATE`, tenantID).Scan(&balance, &reserved); err != nil {
		return TopUpOrder{}, err
	}
	if balance-reserved < amount {
		return TopUpOrder{}, fmt.Errorf("%w: available balance is insufficient to reverse the orphaned credit", ErrCannotResolveTopUp)
	}
	newBalance := balance - amount
	if _, err := tx.Exec(ctx, `UPDATE tenant_wallets SET balance_micros=$2,updated_at=now() WHERE tenant_id=$1`, tenantID, newBalance); err != nil {
		return TopUpOrder{}, err
	}
	if _, err := tx.Exec(ctx, `INSERT INTO billing_ledger
		(tenant_id,currency,amount_micros,balance_after_micros,kind,source_type,source_id,description)
		VALUES ($1,$2,$3,$4,'adjustment','stripe_reconciliation',$5,$6)`,
		tenantID, currency, -amount, newBalance, orderID, reason); err != nil {
		return TopUpOrder{}, err
	}
	if _, err := tx.Exec(ctx, `INSERT INTO billing_reconciliation_resolutions
		(topup_order_id,tenant_id,actor_id,actor_type,reason) VALUES ($1,$2,$3,$4,$5)`,
		orderID, tenantID, actor.ID, actor.Type, reason); err != nil {
		return TopUpOrder{}, err
	}
	if _, err := tx.Exec(ctx, `UPDATE topup_orders SET status='reversed',reconciliation_status='resolved',
		reconciled_at=now(),reconciliation_error=$2 WHERE id=$1`, orderID, reason); err != nil {
		return TopUpOrder{}, err
	}
	if err := tx.Commit(ctx); err != nil {
		return TopUpOrder{}, err
	}
	return s.GetTopUpOrder(ctx, tenantID, orderID)
}

func (s *Service) CreateRefund(ctx context.Context, tenantID, orderID string, input RefundInput) (Refund, error) {
	if !s.stripeEnabled {
		return Refund{}, ErrStripeDisabled
	}
	input.Reason = strings.TrimSpace(input.Reason)
	if input.Reason == "" {
		input.Reason = "requested_by_customer"
	}
	if input.Reason != "requested_by_customer" && input.Reason != "duplicate" && input.Reason != "fraudulent" {
		return Refund{}, errors.New("refund reason must be requested_by_customer, duplicate, or fraudulent")
	}
	tx, err := s.db.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.ReadCommitted})
	if err != nil {
		return Refund{}, err
	}
	defer tx.Rollback(ctx)
	var orderTenant, currency, status, paymentIntentID string
	var orderAmount, refunded int64
	if err := tx.QueryRow(ctx, `SELECT tenant_id::text,currency,status,COALESCE(stripe_payment_intent_id,''),amount_micros,refunded_micros
		FROM topup_orders WHERE id=$1 AND tenant_id=$2 FOR UPDATE`, orderID, tenantID).Scan(&orderTenant, &currency, &status, &paymentIntentID, &orderAmount, &refunded); err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return Refund{}, ErrTopUpOrderNotFound
		}
		return Refund{}, err
	}
	if status != "paid" && status != "partially_refunded" {
		return Refund{}, errors.New("only a paid top-up can be refunded")
	}
	if paymentIntentID == "" {
		return Refund{}, errors.New("top-up has no Stripe PaymentIntent")
	}
	amountMicros, err := minorToMicros(currency, input.AmountMinor)
	if err != nil {
		return Refund{}, ErrInvalidAmount
	}
	var pendingRefunds int64
	if err := tx.QueryRow(ctx, `SELECT COALESCE(sum(amount_micros),0) FROM stripe_refunds
		WHERE topup_order_id=$1 AND status IN ('queued','submitting','pending','requires_action','succeeded')`, orderID).Scan(&pendingRefunds); err != nil {
		return Refund{}, err
	}
	// Legacy successful refunds are already reflected in refunded_micros. New
	// rows are included in the sum, so use the larger value without double-counting.
	committedRefunds := refunded
	if pendingRefunds > committedRefunds {
		committedRefunds = pendingRefunds
	}
	if amountMicros > orderAmount-committedRefunds {
		return Refund{}, ErrInvalidAmount
	}
	var balance, held int64
	if err := tx.QueryRow(ctx, `SELECT balance_micros,reserved_micros FROM tenant_wallets WHERE tenant_id=$1 FOR UPDATE`, tenantID).Scan(&balance, &held); err != nil {
		return Refund{}, err
	}
	available := balance - held
	if available < 0 {
		available = 0
	}
	refundHold := amountMicros
	if refundHold > available {
		refundHold = available
	}
	if _, err := tx.Exec(ctx, `UPDATE tenant_wallets SET reserved_micros=reserved_micros+$2,updated_at=now() WHERE tenant_id=$1`, tenantID, refundHold); err != nil {
		return Refund{}, err
	}
	var result Refund
	err = tx.QueryRow(ctx, `INSERT INTO stripe_refunds (tenant_id,topup_order_id,amount_minor,amount_micros,held_micros,currency,reason)
		VALUES ($1,$2,$3,$4,$5,$6,$7) RETURNING id::text,tenant_id::text,topup_order_id::text,amount_minor,amount_micros,currency,reason,status,last_error,created_at,completed_at`,
		tenantID, orderID, input.AmountMinor, amountMicros, refundHold, currency, input.Reason).Scan(&result.ID, &result.TenantID, &result.TopUpOrderID,
		&result.AmountMinor, &result.AmountMicros, &result.Currency, &result.Reason, &result.Status, &result.LastError, &result.CreatedAt, &result.CompletedAt)
	if err != nil {
		return Refund{}, err
	}
	if err := tx.Commit(ctx); err != nil {
		return Refund{}, err
	}
	return result, nil
}

func (s *Service) ListRefunds(ctx context.Context, tenantID string, limit int) ([]Refund, error) {
	if limit < 1 || limit > 500 {
		limit = 100
	}
	query := `SELECT id::text,tenant_id::text,topup_order_id::text,COALESCE(stripe_refund_id,''),amount_minor,amount_micros,
		currency,reason,status,last_error,created_at,completed_at FROM stripe_refunds`
	args := []any{}
	if tenantID != "" {
		query += ` WHERE tenant_id=$1 ORDER BY created_at DESC LIMIT $2`
		args = []any{tenantID, limit}
	} else {
		query += ` ORDER BY created_at DESC LIMIT $1`
		args = []any{limit}
	}
	rows, err := s.db.Query(ctx, query, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	result := make([]Refund, 0)
	for rows.Next() {
		var item Refund
		if err := rows.Scan(&item.ID, &item.TenantID, &item.TopUpOrderID, &item.StripeRefundID, &item.AmountMinor, &item.AmountMicros, &item.Currency, &item.Reason, &item.Status, &item.LastError, &item.CreatedAt, &item.CompletedAt); err != nil {
			return nil, err
		}
		result = append(result, item)
	}
	return result, rows.Err()
}

func (s *Service) ListDisputes(ctx context.Context, tenantID string, limit int) ([]PaymentDispute, error) {
	if limit < 1 || limit > 500 {
		limit = 100
	}
	query := `SELECT stripe_dispute_id,COALESCE(tenant_id::text,''),COALESCE(topup_order_id::text,''),
		amount_minor,amount_micros,currency,status,reason,debited_micros,uncollected_micros,due_by,updated_at FROM stripe_disputes`
	args := []any{}
	if tenantID != "" {
		query += ` WHERE tenant_id=$1 ORDER BY updated_at DESC LIMIT $2`
		args = []any{tenantID, limit}
	} else {
		query += ` ORDER BY updated_at DESC LIMIT $1`
		args = []any{limit}
	}
	rows, err := s.db.Query(ctx, query, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	result := []PaymentDispute{}
	for rows.Next() {
		var item PaymentDispute
		if err := rows.Scan(&item.ID, &item.TenantID, &item.TopUpOrderID, &item.AmountMinor, &item.AmountMicros, &item.Currency, &item.Status, &item.Reason, &item.DebitedMicros, &item.UncollectedMicros, &item.DueBy, &item.UpdatedAt); err != nil {
			return nil, err
		}
		result = append(result, item)
	}
	return result, rows.Err()
}

func (s *Service) ListInvoices(ctx context.Context, tenantID string, limit int) ([]Invoice, error) {
	if limit < 1 || limit > 500 {
		limit = 100
	}
	query := `SELECT stripe_invoice_id,COALESCE(tenant_id::text,''),COALESCE(topup_order_id::text,''),status,currency,
		amount_due_minor,amount_paid_minor,attempt_count,next_payment_attempt,hosted_invoice_url,invoice_pdf_url,last_failure,updated_at FROM stripe_invoices`
	args := []any{}
	if tenantID != "" {
		query += ` WHERE tenant_id=$1 ORDER BY updated_at DESC LIMIT $2`
		args = []any{tenantID, limit}
	} else {
		query += ` ORDER BY updated_at DESC LIMIT $1`
		args = []any{limit}
	}
	rows, err := s.db.Query(ctx, query, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	result := []Invoice{}
	for rows.Next() {
		var item Invoice
		if err := rows.Scan(&item.ID, &item.TenantID, &item.TopUpOrderID, &item.Status, &item.Currency, &item.AmountDueMinor, &item.AmountPaidMinor, &item.AttemptCount, &item.NextPaymentAttempt, &item.HostedInvoiceURL, &item.InvoicePDFURL, &item.LastFailure, &item.UpdatedAt); err != nil {
			return nil, err
		}
		result = append(result, item)
	}
	return result, rows.Err()
}

func (s *Service) RunStripeOperations(ctx context.Context) {
	if !s.stripeEnabled || s.stripeClient == nil {
		return
	}
	ticker := time.NewTicker(2 * time.Second)
	reconcile := time.NewTicker(30 * time.Minute)
	metrics := time.NewTicker(15 * time.Second)
	defer ticker.Stop()
	defer reconcile.Stop()
	defer metrics.Stop()
	initialCtx, cancel := context.WithTimeout(ctx, 2*time.Minute)
	_, _ = s.Reconcile(initialCtx, 100)
	cancel()
	s.refreshOperationalMetrics(ctx)
	for {
		for i := 0; i < 8; i++ {
			ok, _ := s.processRefundOperation(ctx)
			if !ok {
				break
			}
		}
		_, _ = s.pollPendingRefund(ctx)
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		case <-reconcile.C:
			_, _ = s.Reconcile(ctx, 100)
		case <-metrics.C:
			s.refreshOperationalMetrics(ctx)
		}
	}
}

func (s *Service) refreshOperationalMetrics(ctx context.Context) {
	if s.metrics == nil {
		return
	}
	status, err := s.OperationalStatus(ctx)
	if err == nil {
		s.metrics.SetStripeOperations(status)
	}
}

func (s *Service) OperationalStatus(ctx context.Context) (OperationalStatus, error) {
	status := OperationalStatus{StripeEnabled: s.stripeEnabled, ReconciliationStatus: "disabled"}
	if err := s.db.QueryRow(ctx, `SELECT
		count(*) FILTER (WHERE status IN ('queued','submitting','pending','requires_action')),
		min(created_at) FILTER (WHERE status IN ('queued','submitting','pending','requires_action')),
		COALESCE((SELECT sum(uncollected_micros) FROM stripe_refunds),0)+
		COALESCE((SELECT sum(uncollected_micros) FROM stripe_disputes),0)+
		COALESCE((SELECT sum(uncollected_micros) FROM usage_events),0)
		FROM stripe_refunds`).Scan(&status.RefundBacklog, &status.OldestRefund, &status.UncollectedMicros); err != nil {
		return status, err
	}
	if err := s.db.QueryRow(ctx, `SELECT count(*),min(created_at) FROM stripe_webhook_events
		WHERE processed_at IS NULL`).Scan(&status.UnprocessedWebhooks, &status.OldestUnprocessedWebhook); err != nil {
		return status, err
	}
	if err := s.db.QueryRow(ctx, `SELECT count(*) FROM usage_events WHERE metering_status='missing'`).Scan(&status.UnmeteredSuccesses); err != nil {
		return status, err
	}
	if !s.stripeEnabled {
		return status, nil
	}
	err := s.db.QueryRow(ctx, `SELECT status,mismatch_count,completed_at,error FROM billing_reconciliation_runs
		WHERE status<>'running' ORDER BY started_at DESC LIMIT 1`).Scan(&status.ReconciliationStatus,
		&status.ReconciliationMismatches, &status.ReconciliationCompletedAt, &status.ReconciliationError)
	if errors.Is(err, pgx.ErrNoRows) {
		status.ReconciliationStatus = "never_run"
		return status, nil
	}
	return status, err
}

func (s *Service) processRefundOperation(ctx context.Context) (bool, error) {
	tx, err := s.db.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.ReadCommitted})
	if err != nil {
		return false, err
	}
	defer tx.Rollback(ctx)
	var id, orderID, paymentIntentID, reason string
	var amount int64
	err = tx.QueryRow(ctx, `WITH selected AS (
		SELECT r.id FROM stripe_refunds r WHERE r.available_at<=now() AND
			(r.status='queued' OR (r.status='submitting' AND r.updated_at<now()-interval '5 minutes'))
		ORDER BY r.available_at,r.created_at FOR UPDATE SKIP LOCKED LIMIT 1)
		UPDATE stripe_refunds r SET status='submitting',attempts=attempts+1,updated_at=now()
		FROM selected,topup_orders o WHERE r.id=selected.id AND o.id=r.topup_order_id
		RETURNING r.id::text,r.topup_order_id::text,o.stripe_payment_intent_id,r.amount_minor,r.reason`).Scan(&id, &orderID, &paymentIntentID, &amount, &reason)
	if errors.Is(err, pgx.ErrNoRows) {
		return false, tx.Commit(ctx)
	}
	if err != nil {
		return false, err
	}
	if err := tx.Commit(ctx); err != nil {
		return false, err
	}
	params := &stripe.RefundCreateParams{Amount: stripe.Int64(amount), PaymentIntent: stripe.String(paymentIntentID), Reason: stripe.String(reason),
		Metadata: map[string]string{"aigw_refund_id": id, "aigw_topup_order_id": orderID}}
	params.SetIdempotencyKey("aigw_refund_" + id)
	refund, err := s.stripeClient.V1Refunds.Create(ctx, params)
	if err != nil {
		message := err.Error()
		if len(message) > 1000 {
			message = message[:1000]
		}
		_, _ = s.db.Exec(ctx, `UPDATE stripe_refunds SET status='queued',last_error=$2,
			available_at=now()+make_interval(secs=>LEAST(1800,power(2,LEAST(attempts,10))::int)),updated_at=now() WHERE id=$1`, id, message)
		return true, err
	}
	return true, s.applyRefund(ctx, refund)
}

func (s *Service) pollPendingRefund(ctx context.Context) (bool, error) {
	var id string
	err := s.db.QueryRow(ctx, `UPDATE stripe_refunds SET available_at=now()+interval '1 minute',updated_at=now()
		WHERE id=(SELECT id FROM stripe_refunds WHERE status IN ('pending','requires_action')
			AND stripe_refund_id IS NOT NULL AND available_at<=now() ORDER BY available_at LIMIT 1 FOR UPDATE SKIP LOCKED)
		RETURNING stripe_refund_id`).Scan(&id)
	if errors.Is(err, pgx.ErrNoRows) {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	refund, err := s.stripeClient.V1Refunds.Retrieve(ctx, id, &stripe.RefundRetrieveParams{})
	if err != nil {
		return true, err
	}
	return true, s.applyRefund(ctx, refund)
}

func (s *Service) applyRefund(ctx context.Context, refund *stripe.Refund) error {
	if refund == nil || refund.ID == "" {
		return ErrInvalidAmount
	}
	tx, err := s.db.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.ReadCommitted})
	if err != nil {
		return err
	}
	defer tx.Rollback(ctx)
	if err := s.applyRefundTx(ctx, tx, refund); err != nil {
		return err
	}
	return tx.Commit(ctx)
}

func (s *Service) applyRefundTx(ctx context.Context, tx pgx.Tx, refund *stripe.Refund) error {
	if refund == nil || refund.ID == "" {
		return ErrInvalidAmount
	}
	localID := refund.Metadata["aigw_refund_id"]
	var id, tenantID, orderID, currentStatus string
	var amountMicros, held int64
	query := `SELECT id::text,tenant_id::text,topup_order_id::text,amount_micros,held_micros,status FROM stripe_refunds WHERE `
	arg := refund.ID
	if localID != "" {
		query += `id=$1 FOR UPDATE`
		arg = localID
	} else {
		query += `stripe_refund_id=$1 FOR UPDATE`
	}
	err := tx.QueryRow(ctx, query, arg).Scan(&id, &tenantID, &orderID, &amountMicros, &held, &currentStatus)
	if errors.Is(err, pgx.ErrNoRows) {
		paymentIntentID := ""
		if refund.PaymentIntent != nil {
			paymentIntentID = refund.PaymentIntent.ID
		}
		if paymentIntentID == "" {
			return ErrInvalidAmount
		}
		var currency string
		if err := tx.QueryRow(ctx, `SELECT id::text,tenant_id::text,currency FROM topup_orders WHERE stripe_payment_intent_id=$1 FOR UPDATE`, paymentIntentID).Scan(&orderID, &tenantID, &currency); err != nil {
			return err
		}
		amountMicros, err = minorToMicros(currency, refund.Amount)
		if err != nil {
			return err
		}
		err = tx.QueryRow(ctx, `INSERT INTO stripe_refunds (tenant_id,topup_order_id,stripe_refund_id,amount_minor,amount_micros,currency,reason,status)
			VALUES ($1,$2,$3,$4,$5,$6,$7,$8) RETURNING id::text`, tenantID, orderID, refund.ID, refund.Amount, amountMicros, currency, string(refund.Reason), string(refund.Status)).Scan(&id)
		if err != nil {
			return err
		}
		currentStatus = ""
	} else if err != nil {
		return err
	}
	if _, err := tx.Exec(ctx, `UPDATE stripe_refunds SET stripe_refund_id=$2,status=$3,last_error=$4,updated_at=now(),
		completed_at=CASE WHEN $3 IN ('succeeded','failed','canceled') THEN now() ELSE completed_at END WHERE id=$1`, id, refund.ID, string(refund.Status), string(refund.FailureReason)); err != nil {
		return err
	}
	if refund.Status == stripe.RefundStatusSucceeded && currentStatus != "succeeded" {
		var balance, reserved int64
		if err := tx.QueryRow(ctx, `SELECT balance_micros,reserved_micros FROM tenant_wallets WHERE tenant_id=$1 FOR UPDATE`, tenantID).Scan(&balance, &reserved); err != nil {
			return err
		}
		if held > 0 {
			if held > reserved {
				return errors.New("refund hold invariant violated")
			}
			reserved -= held
		}
		debit := amountMicros
		if debit > balance-reserved {
			debit = balance - reserved
		}
		if debit < 0 {
			debit = 0
		}
		newBalance := balance - debit
		if _, err := tx.Exec(ctx, `UPDATE tenant_wallets SET balance_micros=$2,reserved_micros=$3,updated_at=now() WHERE tenant_id=$1`, tenantID, newBalance, reserved); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `UPDATE stripe_refunds SET held_micros=0,uncollected_micros=$2 WHERE id=$1`, id, amountMicros-debit); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `INSERT INTO billing_ledger (tenant_id,currency,amount_micros,balance_after_micros,kind,source_type,source_id,description)
			SELECT $1,currency,$2,$3,'refund','stripe_refund',$4,'Stripe top-up refund' FROM topup_orders WHERE id=$5
			ON CONFLICT (source_type,source_id) DO NOTHING`, tenantID, -debit, newBalance, refund.ID, orderID); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `UPDATE topup_orders SET refunded_micros=LEAST(amount_micros,refunded_micros+$2),
			status=CASE WHEN refunded_micros+$2>=amount_micros THEN 'refunded' ELSE 'partially_refunded' END WHERE id=$1`, orderID, amountMicros); err != nil {
			return err
		}
	} else if (refund.Status == stripe.RefundStatusFailed || refund.Status == stripe.RefundStatusCanceled) && held > 0 {
		if _, err := tx.Exec(ctx, `UPDATE tenant_wallets SET reserved_micros=reserved_micros-$2,updated_at=now() WHERE tenant_id=$1`, tenantID, held); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `UPDATE stripe_refunds SET held_micros=0 WHERE id=$1`, id); err != nil {
			return err
		}
	}
	return nil
}

func (s *Service) applyDisputeTx(ctx context.Context, tx pgx.Tx, dispute *stripe.Dispute, eventType stripe.EventType) error {
	paymentIntentID := ""
	if dispute.PaymentIntent != nil {
		paymentIntentID = dispute.PaymentIntent.ID
	}
	if paymentIntentID == "" && dispute.Charge != nil && dispute.Charge.PaymentIntent != nil {
		paymentIntentID = dispute.Charge.PaymentIntent.ID
	}
	if paymentIntentID == "" {
		return ErrInvalidAmount
	}
	var orderID, tenantID, currency string
	if err := tx.QueryRow(ctx, `SELECT id::text,tenant_id::text,currency FROM topup_orders WHERE stripe_payment_intent_id=$1 FOR UPDATE`, paymentIntentID).Scan(&orderID, &tenantID, &currency); err != nil {
		return err
	}
	amountMicros, err := minorToMicros(currency, dispute.Amount)
	if err != nil {
		return err
	}
	dueBy := (*time.Time)(nil)
	if dispute.EvidenceDetails != nil && dispute.EvidenceDetails.DueBy > 0 {
		value := time.Unix(dispute.EvidenceDetails.DueBy, 0).UTC()
		dueBy = &value
	}
	var previousDebited int64
	queryErr := tx.QueryRow(ctx, `SELECT debited_micros FROM stripe_disputes WHERE stripe_dispute_id=$1 FOR UPDATE`, dispute.ID).Scan(&previousDebited)
	if queryErr != nil && !errors.Is(queryErr, pgx.ErrNoRows) {
		return queryErr
	}
	if _, err := tx.Exec(ctx, `INSERT INTO stripe_disputes (stripe_dispute_id,tenant_id,topup_order_id,stripe_payment_intent_id,
		amount_minor,amount_micros,currency,status,reason,due_by) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)
		ON CONFLICT (stripe_dispute_id) DO UPDATE SET status=EXCLUDED.status,reason=EXCLUDED.reason,
		due_by=EXCLUDED.due_by,updated_at=now(),closed_at=CASE WHEN EXCLUDED.status IN ('won','lost') THEN now() ELSE stripe_disputes.closed_at END`,
		dispute.ID, tenantID, orderID, paymentIntentID, dispute.Amount, amountMicros, currency, string(dispute.Status), string(dispute.Reason), dueBy); err != nil {
		return err
	}
	shouldDebit := eventType == stripe.EventTypeChargeDisputeCreated || eventType == stripe.EventTypeChargeDisputeFundsWithdrawn
	shouldReverse := eventType == stripe.EventTypeChargeDisputeFundsReinstated || dispute.Status == stripe.DisputeStatusWon
	if shouldDebit && previousDebited == 0 {
		var balance, reserved int64
		if err := tx.QueryRow(ctx, `SELECT balance_micros,reserved_micros FROM tenant_wallets WHERE tenant_id=$1 FOR UPDATE`, tenantID).Scan(&balance, &reserved); err != nil {
			return err
		}
		debit := amountMicros
		if debit > balance-reserved {
			debit = balance - reserved
		}
		if debit < 0 {
			debit = 0
		}
		newBalance := balance - debit
		if _, err := tx.Exec(ctx, `UPDATE tenant_wallets SET balance_micros=$2,updated_at=now() WHERE tenant_id=$1`, tenantID, newBalance); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `UPDATE stripe_disputes SET debited_micros=$2,uncollected_micros=$3 WHERE stripe_dispute_id=$1`, dispute.ID, debit, amountMicros-debit); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `INSERT INTO billing_ledger (tenant_id,currency,amount_micros,balance_after_micros,kind,source_type,source_id,description)
			VALUES ($1,$2,$3,$4,'dispute','stripe_dispute',$5,'Stripe payment dispute') ON CONFLICT DO NOTHING`, tenantID, currency, -debit, newBalance, dispute.ID); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `UPDATE topup_orders SET disputed_micros=GREATEST(disputed_micros,$2),status='disputed' WHERE id=$1`, orderID, amountMicros); err != nil {
			return err
		}
	} else if shouldReverse && previousDebited > 0 {
		var balance int64
		if err := tx.QueryRow(ctx, `SELECT balance_micros FROM tenant_wallets WHERE tenant_id=$1 FOR UPDATE`, tenantID).Scan(&balance); err != nil {
			return err
		}
		newBalance := balance + previousDebited
		if _, err := tx.Exec(ctx, `UPDATE tenant_wallets SET balance_micros=$2,updated_at=now() WHERE tenant_id=$1`, tenantID, newBalance); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `UPDATE stripe_disputes SET debited_micros=0,uncollected_micros=0 WHERE stripe_dispute_id=$1`, dispute.ID); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `INSERT INTO billing_ledger (tenant_id,currency,amount_micros,balance_after_micros,kind,source_type,source_id,description)
			VALUES ($1,$2,$3,$4,'dispute_reversal','stripe_dispute_reversal',$5,'Stripe dispute funds reinstated') ON CONFLICT DO NOTHING`, tenantID, currency, previousDebited, newBalance, dispute.ID); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `UPDATE topup_orders SET disputed_micros=0,status=CASE WHEN refunded_micros=0 THEN 'paid' WHEN refunded_micros<amount_micros THEN 'partially_refunded' ELSE 'refunded' END WHERE id=$1`, orderID); err != nil {
			return err
		}
	}
	return nil
}

func (s *Service) applyInvoiceTx(ctx context.Context, tx pgx.Tx, invoice *stripe.Invoice) error {
	orderID := invoice.Metadata["aigw_topup_order_id"]
	tenantID := invoice.Metadata["aigw_tenant_id"]
	customerID := ""
	if invoice.Customer != nil {
		customerID = invoice.Customer.ID
	}
	if tenantID == "" && customerID != "" {
		_ = tx.QueryRow(ctx, `SELECT tenant_id::text FROM stripe_customers WHERE stripe_customer_id=$1`, customerID).Scan(&tenantID)
	}
	if orderID == "" {
		_ = tx.QueryRow(ctx, `SELECT id::text FROM topup_orders WHERE stripe_invoice_id=$1`, invoice.ID).Scan(&orderID)
	}
	failure := ""
	if invoice.LastFinalizationError != nil {
		failure = invoice.LastFinalizationError.Msg
	}
	var nextAttempt *time.Time
	if invoice.NextPaymentAttempt > 0 {
		value := time.Unix(invoice.NextPaymentAttempt, 0).UTC()
		nextAttempt = &value
	}
	if _, err := tx.Exec(ctx, `INSERT INTO stripe_invoices (stripe_invoice_id,tenant_id,topup_order_id,stripe_customer_id,status,currency,
		amount_due_minor,amount_paid_minor,attempt_count,next_payment_attempt,hosted_invoice_url,invoice_pdf_url,last_failure)
		VALUES ($1,NULLIF($2,'')::uuid,NULLIF($3,'')::uuid,NULLIF($4,''),$5,$6,$7,$8,$9,$10,$11,$12,$13)
		ON CONFLICT (stripe_invoice_id) DO UPDATE SET status=EXCLUDED.status,amount_due_minor=EXCLUDED.amount_due_minor,
		amount_paid_minor=EXCLUDED.amount_paid_minor,attempt_count=EXCLUDED.attempt_count,next_payment_attempt=EXCLUDED.next_payment_attempt,
		hosted_invoice_url=EXCLUDED.hosted_invoice_url,invoice_pdf_url=EXCLUDED.invoice_pdf_url,last_failure=EXCLUDED.last_failure,updated_at=now()`,
		invoice.ID, tenantID, orderID, customerID, string(invoice.Status), string(invoice.Currency), invoice.AmountDue, invoice.AmountPaid,
		invoice.AttemptCount, nextAttempt, invoice.HostedInvoiceURL, invoice.InvoicePDF, failure); err != nil {
		return err
	}
	if orderID != "" {
		_, err := tx.Exec(ctx, `UPDATE topup_orders SET stripe_invoice_id=$2,invoice_url=COALESCE(NULLIF($3,''),invoice_url),
		invoice_pdf_url=COALESCE(NULLIF($4,''),invoice_pdf_url) WHERE id=$1`, orderID, invoice.ID, invoice.HostedInvoiceURL, invoice.InvoicePDF)
		return err
	}
	return nil
}

func (s *Service) Reconcile(ctx context.Context, limit int) (ReconciliationResult, error) {
	if !s.stripeEnabled || s.stripeClient == nil {
		return ReconciliationResult{}, ErrStripeDisabled
	}
	if limit < 1 || limit > 500 {
		limit = 100
	}
	var result ReconciliationResult
	if err := s.db.QueryRow(ctx, `INSERT INTO billing_reconciliation_runs (status) VALUES ('running') RETURNING id::text`).Scan(&result.ID); err != nil {
		return result, err
	}
	rows, err := s.db.Query(ctx, `SELECT id::text,stripe_session_id,status,amount_minor,currency,reconciliation_status FROM topup_orders
		WHERE stripe_session_id IS NOT NULL ORDER BY created_at DESC LIMIT $1`, limit)
	if err != nil {
		return s.failReconciliation(ctx, result, err)
	}
	type order struct {
		id, session, status, currency, reconciliationStatus string
		amount                                              int64
	}
	var orders []order
	for rows.Next() {
		var item order
		if err := rows.Scan(&item.id, &item.session, &item.status, &item.amount, &item.currency, &item.reconciliationStatus); err != nil {
			rows.Close()
			return s.failReconciliation(ctx, result, err)
		}
		orders = append(orders, item)
	}
	rows.Close()
	for _, item := range orders {
		session, retrieveErr := s.stripeClient.V1CheckoutSessions.Retrieve(ctx, item.session, &stripe.CheckoutSessionRetrieveParams{})
		result.CheckedOrders++
		if retrieveErr != nil {
			if stripeResourceMissing(retrieveErr) && item.reconciliationStatus == "resolved" && (item.status == "failed" || item.status == "reversed") {
				continue
			}
			message := truncateError(retrieveErr)
			typeName := "stripe_retrieve_failed"
			state := "mismatch"
			if stripeResourceMissing(retrieveErr) {
				typeName = "stripe_session_missing"
				state = "missing"
			}
			if err := s.updateOrderReconciliation(ctx, item.id, state, message); err != nil {
				return s.failReconciliation(ctx, result, fmt.Errorf("record reconciliation retrieval failure for order %s: %w", item.id, err))
			}
			result.Mismatches = append(result.Mismatches, map[string]any{"order_id": item.id, "type": typeName, "error": message})
			continue
		}
		expected := item.status == "paid" || item.status == "partially_refunded" || item.status == "refunded" || item.status == "disputed"
		stripePaid := session.PaymentStatus == stripe.CheckoutSessionPaymentStatusPaid
		if session.AmountTotal != item.amount || string(session.Currency) != item.currency {
			result.Mismatches = append(result.Mismatches, map[string]any{"order_id": item.id, "type": "checkout_mismatch", "local_status": item.status, "stripe_payment_status": session.PaymentStatus, "local_amount": item.amount, "stripe_amount": session.AmountTotal})
			if err := s.updateOrderReconciliation(ctx, item.id, "mismatch", "amount or currency mismatch"); err != nil {
				return s.failReconciliation(ctx, result, fmt.Errorf("record checkout mismatch for order %s: %w", item.id, err))
			}
			continue
		}
		if stripePaid && !expected {
			raw, marshalErr := json.Marshal(session)
			if marshalErr != nil {
				return s.failReconciliation(ctx, result, fmt.Errorf("encode Stripe Checkout Session for order %s: %w", item.id, marshalErr))
			}
			if repairErr := s.processStripeEvent(ctx, stripe.Event{ID: "reconcile_" + session.ID + "_paid", Type: stripe.EventTypeCheckoutSessionCompleted, Data: &stripe.EventData{Raw: raw}}); repairErr != nil {
				message := truncateError(repairErr)
				result.Mismatches = append(result.Mismatches, map[string]any{"order_id": item.id, "type": "checkout_repair_failed", "error": message})
				if err := s.updateOrderReconciliation(ctx, item.id, "mismatch", message); err != nil {
					return s.failReconciliation(ctx, result, fmt.Errorf("record checkout repair failure for order %s: %w", item.id, err))
				}
				continue
			}
			result.Repairs = append(result.Repairs, map[string]any{"order_id": item.id, "type": "credited_paid_checkout"})
			if err := s.updateOrderReconciliation(ctx, item.id, "repaired", ""); err != nil {
				return s.failReconciliation(ctx, result, fmt.Errorf("record checkout repair for order %s: %w", item.id, err))
			}
			continue
		}
		if expected != stripePaid {
			result.Mismatches = append(result.Mismatches, map[string]any{"order_id": item.id, "type": "checkout_payment_state_mismatch", "local_status": item.status, "stripe_payment_status": session.PaymentStatus})
			if err := s.updateOrderReconciliation(ctx, item.id, "mismatch", "payment state mismatch"); err != nil {
				return s.failReconciliation(ctx, result, fmt.Errorf("record payment-state mismatch for order %s: %w", item.id, err))
			}
			continue
		}
		if err := s.updateOrderReconciliation(ctx, item.id, "ok", ""); err != nil {
			return s.failReconciliation(ctx, result, fmt.Errorf("record clean reconciliation for order %s: %w", item.id, err))
		}
	}
	result.MismatchCount = int64(len(result.Mismatches))
	result.Status = "clean"
	if result.MismatchCount > 0 {
		result.Status = "mismatch"
	}
	report, marshalErr := json.Marshal(result.Mismatches)
	if marshalErr != nil {
		return s.failReconciliation(ctx, result, fmt.Errorf("encode reconciliation report: %w", marshalErr))
	}
	_, err = s.db.Exec(ctx, `UPDATE billing_reconciliation_runs SET status=$2,checked_orders=$3,mismatch_count=$4,report=$5,completed_at=now() WHERE id=$1`, result.ID, result.Status, result.CheckedOrders, result.MismatchCount, report)
	if err != nil {
		return s.failReconciliation(ctx, result, fmt.Errorf("complete reconciliation run: %w", err))
	}
	return result, nil
}

func (s *Service) updateOrderReconciliation(ctx context.Context, orderID, status, message string) error {
	command, err := s.db.Exec(ctx, `UPDATE topup_orders SET reconciliation_status=$2,reconciled_at=now(),reconciliation_error=$3 WHERE id=$1`, orderID, status, message)
	if err != nil {
		return err
	}
	if command.RowsAffected() != 1 {
		return fmt.Errorf("expected one top-up order, updated %d", command.RowsAffected())
	}
	return nil
}

func stripeResourceMissing(err error) bool {
	var stripeErr *stripe.Error
	return errors.As(err, &stripeErr) && (stripeErr.Code == stripe.ErrorCodeResourceMissing || stripeErr.HTTPStatusCode == 404)
}

func truncateError(err error) string {
	if err == nil {
		return ""
	}
	message := err.Error()
	if len(message) > 1000 {
		return message[:1000]
	}
	return message
}

func (s *Service) failReconciliation(ctx context.Context, result ReconciliationResult, cause error) (ReconciliationResult, error) {
	if _, err := s.db.Exec(ctx, `UPDATE billing_reconciliation_runs SET status='failed',error=$2,completed_at=now() WHERE id=$1`, result.ID, truncateError(cause)); err != nil {
		return result, errors.Join(cause, fmt.Errorf("record failed reconciliation run: %w", err))
	}
	return result, cause
}

func (s *Service) WriteFinancialCSV(ctx context.Context, tenantID string, output io.Writer) error {
	query := `
		SELECT id::text, tenant_id::text, COALESCE(project_id::text, ''), currency, amount_micros,
		       balance_after_micros, kind, source_type, source_id, description, created_at
		FROM billing_ledger`
	args := []any{}
	if strings.TrimSpace(tenantID) != "" {
		query += ` WHERE tenant_id=$1`
		args = append(args, tenantID)
	}
	query += ` ORDER BY created_at, id`
	rows, err := s.db.Query(ctx, query, args...)
	if err != nil {
		return err
	}
	defer rows.Close()
	w := csv.NewWriter(output)
	if err := w.Write([]string{"timestamp", "tenant_id", "project_id", "currency", "kind", "amount_micros", "balance_after_micros", "source_type", "source_id", "description"}); err != nil {
		return err
	}
	for rows.Next() {
		var e LedgerEntry
		if err := rows.Scan(&e.ID, &e.TenantID, &e.ProjectID, &e.Currency, &e.AmountMicros,
			&e.BalanceAfterMicros, &e.Kind, &e.SourceType, &e.SourceID, &e.Description, &e.CreatedAt); err != nil {
			return err
		}
		if err := w.Write([]string{e.CreatedAt.UTC().Format(time.RFC3339Nano), e.TenantID, e.ProjectID, e.Currency, e.Kind, strconv.FormatInt(e.AmountMicros, 10), strconv.FormatInt(e.BalanceAfterMicros, 10), e.SourceType, e.SourceID, e.Description}); err != nil {
			return err
		}
	}
	if err := rows.Err(); err != nil {
		return err
	}
	w.Flush()
	return w.Error()
}