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 } customerID, err := s.ensureStripeCustomer(ctx, tenantID) if err != nil { return PortalResult{}, err } if customerID == "" { return PortalResult{}, errors.New("no Stripe customer exists for this account") } 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 < 4; i++ { ok, _ := s.processAutoTopUpOnce(ctx) if !ok { break } } 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)) } } if s.retrieveStripePaymentIntent == nil { return s.failReconciliation(ctx, result, errors.New("Stripe PaymentIntent retrieval is unavailable")) } piRows, err := s.db.Query(ctx, `SELECT id::text,stripe_payment_intent_id,status,amount_minor,currency FROM topup_orders WHERE trigger_type='auto' AND stripe_payment_intent_id IS NOT NULL ORDER BY created_at DESC LIMIT $1`, limit) if err != nil { return s.failReconciliation(ctx, result, err) } type paymentIntentOrder struct { id, paymentIntent, status, currency string amount int64 } var paymentIntentOrders []paymentIntentOrder for piRows.Next() { var item paymentIntentOrder if err := piRows.Scan(&item.id, &item.paymentIntent, &item.status, &item.amount, &item.currency); err != nil { piRows.Close() return s.failReconciliation(ctx, result, err) } paymentIntentOrders = append(paymentIntentOrders, item) } piRows.Close() for _, item := range paymentIntentOrders { intent, retrieveErr := s.retrieveStripePaymentIntent(ctx, item.paymentIntent, &stripe.PaymentIntentRetrieveParams{}) result.CheckedOrders++ if retrieveErr != nil { message := truncateError(retrieveErr) if err := s.updateOrderReconciliation(ctx, item.id, "mismatch", message); err != nil { return s.failReconciliation(ctx, result, fmt.Errorf("record automatic top-up retrieval failure for order %s: %w", item.id, err)) } result.Mismatches = append(result.Mismatches, map[string]any{"order_id": item.id, "type": "payment_intent_retrieve_failed", "error": message}) continue } if intent == nil || intent.Amount != item.amount || string(intent.Currency) != item.currency { if err := s.updateOrderReconciliation(ctx, item.id, "mismatch", "amount or currency mismatch"); err != nil { return s.failReconciliation(ctx, result, err) } result.Mismatches = append(result.Mismatches, map[string]any{"order_id": item.id, "type": "payment_intent_mismatch"}) continue } expectedPaid := item.status == "paid" || item.status == "partially_refunded" || item.status == "refunded" || item.status == "disputed" stripePaid := intent.Status == stripe.PaymentIntentStatusSucceeded if stripePaid && !expectedPaid { if repairErr := s.creditAutoTopUpPaymentIntent(ctx, intent); repairErr != nil { message := truncateError(repairErr) result.Mismatches = append(result.Mismatches, map[string]any{"order_id": item.id, "type": "payment_intent_repair_failed", "error": message}) if err := s.updateOrderReconciliation(ctx, item.id, "mismatch", message); err != nil { return s.failReconciliation(ctx, result, err) } continue } result.Repairs = append(result.Repairs, map[string]any{"order_id": item.id, "type": "credited_paid_payment_intent"}) if err := s.updateOrderReconciliation(ctx, item.id, "repaired", ""); err != nil { return s.failReconciliation(ctx, result, err) } continue } if expectedPaid != stripePaid { result.Mismatches = append(result.Mismatches, map[string]any{"order_id": item.id, "type": "payment_intent_state_mismatch", "local_status": item.status, "stripe_status": intent.Status}) if err := s.updateOrderReconciliation(ctx, item.id, "mismatch", "payment state mismatch"); err != nil { return s.failReconciliation(ctx, result, err) } continue } if err := s.updateOrderReconciliation(ctx, item.id, "ok", ""); err != nil { return s.failReconciliation(ctx, result, 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() }