summaryrefslogtreecommitdiff
path: root/internal/billing/operations.go
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--internal/billing/operations.go881
1 files changed, 881 insertions, 0 deletions
diff --git a/internal/billing/operations.go b/internal/billing/operations.go
new file mode 100644
index 0000000..a6dc653
--- /dev/null
+++ b/internal/billing/operations.go
@@ -0,0 +1,881 @@
+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()
+}
generated by cgit v1.2.3 (git 2.46.0) at 2026-10-03 14:36:59 +0000