From eadb2ffe85c43cf6fc741c9823cd28eedb4a844c Mon Sep 17 00:00:00 2001 From: Chia Date: Wed, 5 Aug 2026 22:01:29 +1200 Subject: feat: harden prepaid billing and commercial operations --- internal/billing/operations.go | 881 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 881 insertions(+) create mode 100644 internal/billing/operations.go (limited to 'internal/billing/operations.go') 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, ¤cy, &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, ¤cy, &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 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, ¤tStatus) + 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, ¤cy); 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, ¤cy); 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 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() +} -- cgit v1.2.3