package server import ( "context" "database/sql" "errors" "fmt" "log/slog" "math" "net/http" "strconv" "strings" "time" "git.coopcloud.tech/wiki-cafe/member-console/internal/billing" "git.coopcloud.tech/wiki-cafe/member-console/internal/integration" stripedb "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/stripe/store" "git.coopcloud.tech/wiki-cafe/member-console/internal/web" ) // BillingAccountViewModel represents a billing account for operator template rendering type BillingAccountViewModel struct { BillingAccountID string OrgID string OrgName string Name string Status string StripeCustomerID string StripeSyncStatus string CreatedAt string } // BillingAccountsData holds data for the billing accounts list partial type BillingAccountsData struct { Accounts []BillingAccountViewModel Error string // StripeConfigured gates the empty-view copy (see // operator_billing_accounts.html): an empty list because Stripe was // never configured (blocked on the operator) must read differently from // an empty list because Stripe is configured but no webhook events have // landed yet (waiting on the world) — the page-level "no events // processed" statement already covers the latter. StripeConfigured bool } // latestProcessedWebhookEventAt returns the most recent time any provider's // webhook event finished processing (processed_at IS NOT NULL): the four // operator billing views are projections from these events, so this is the // recency stamp the honest-surfaces delta requires (design.md Decision 3; // spec: operator-billing-views) — "data as of ..." when at least one event // has ever finished processing, ok=false when none has (a fresh deployment, // or Stripe never configured), so the caller can state that plainly instead // of rendering an empty or stale-looking table as if it were current. func (h *OperatorPartialsHandler) latestProcessedWebhookEventAt(ctx context.Context) (time.Time, bool) { var latest sql.NullTime if err := h.Database.QueryRowContext(ctx, `SELECT MAX(processed_at) FROM core.webhook_events WHERE processed_at IS NOT NULL`, ).Scan(&latest); err != nil { h.Logger.Error("failed to load latest processed webhook event", "error", err) return time.Time{}, false } if !latest.Valid { return time.Time{}, false } return latest.Time, true } // loadBillingAccountsData hydrates the billing-accounts listing. func (h *OperatorPartialsHandler) loadBillingAccountsData(r *http.Request) BillingAccountsData { stripeConfigured, _ := configurationReadiness(h.IntegrationConfigs, "stripe") accounts, err := h.BillingQ.ListBillingAccounts(r.Context()) if err != nil { h.Logger.Error("failed to list billing accounts", "error", err) return BillingAccountsData{Error: "Failed to load billing accounts"} } vms := make([]BillingAccountViewModel, len(accounts)) for i, acc := range accounts { orgName := "" if org, err := h.OrgQ.GetOrganizationByID(r.Context(), acc.OrgID); err == nil { orgName = org.Name } stripeCustomerID := "" syncStatus := "not_mapped" if mapping, err := h.StripeQ.GetCustomerMappingByBillingAccountID(r.Context(), acc.BillingAccountID); err == nil { if mapping.StripeCustomerID.Valid { stripeCustomerID = mapping.StripeCustomerID.String } syncStatus = mapping.SyncStatus } vms[i] = BillingAccountViewModel{ BillingAccountID: acc.BillingAccountID, OrgID: acc.OrgID, OrgName: orgName, Name: acc.Name, Status: acc.Status, StripeCustomerID: stripeCustomerID, StripeSyncStatus: syncStatus, CreatedAt: acc.CreatedAt.Format("Jan 2, 2006"), } } return BillingAccountsData{Accounts: vms, StripeConfigured: stripeConfigured} } // SubscriptionViewModel represents a subscription for operator template rendering type SubscriptionViewModel struct { SubscriptionID string BillingAccountID string BillingAccountName string Status string StatusClass string CurrentPeriodStart string CurrentPeriodEnd string CancelAtPeriodEnd bool StripeSubscriptionID string StripeSyncStatus string CreatedAt string } // SubscriptionsData holds data for the subscriptions list partial type SubscriptionsData struct { Subscriptions []SubscriptionViewModel Error string // StripeConfigured gates the empty-view copy; see BillingAccountsData. StripeConfigured bool } // loadSubscriptionsData hydrates the subscriptions listing. func (h *OperatorPartialsHandler) loadSubscriptionsData(r *http.Request) SubscriptionsData { stripeConfigured, _ := configurationReadiness(h.IntegrationConfigs, "stripe") subs, err := h.BillingQ.ListSubscriptions(r.Context()) if err != nil { h.Logger.Error("failed to list subscriptions", "error", err) return SubscriptionsData{Error: "Failed to load subscriptions"} } vms := make([]SubscriptionViewModel, len(subs)) for i, sub := range subs { stripeSubID := "" syncStatus := "not_mapped" if mapping, err := h.StripeQ.GetSubscriptionMappingBySubscriptionID(r.Context(), sub.SubscriptionID); err == nil { if mapping.StripeSubscriptionID.Valid { stripeSubID = mapping.StripeSubscriptionID.String } syncStatus = mapping.SyncStatus } periodStart := "" if sub.CurrentPeriodStart.Valid { periodStart = sub.CurrentPeriodStart.Time.Format("Jan 2, 2006") } periodEnd := "" if sub.CurrentPeriodEnd.Valid { periodEnd = sub.CurrentPeriodEnd.Time.Format("Jan 2, 2006") } vms[i] = SubscriptionViewModel{ SubscriptionID: sub.SubscriptionID, BillingAccountID: sub.BillingAccountID, BillingAccountName: sub.BillingAccountName, Status: sub.Status, StatusClass: getStatusBadgeClass(sub.Status), CurrentPeriodStart: periodStart, CurrentPeriodEnd: periodEnd, CancelAtPeriodEnd: sub.CancelAtPeriodEnd, StripeSubscriptionID: stripeSubID, StripeSyncStatus: syncStatus, CreatedAt: sub.CreatedAt.Format("Jan 2, 2006"), } } return SubscriptionsData{Subscriptions: vms, StripeConfigured: stripeConfigured} } // InvoiceViewModel represents an invoice for operator template rendering type InvoiceViewModel struct { InvoiceID string BillingAccountID string BillingAccountName string Status string StatusClass string AmountDue string AmountPaid string Currency string DueDate string PaidAt string StripeInvoiceID string StripeSyncStatus string CreatedAt string } // InvoicesData holds data for the invoices list partial type InvoicesData struct { Invoices []InvoiceViewModel Error string // StripeConfigured gates the empty-view copy; see BillingAccountsData. StripeConfigured bool } // zeroDecimalCurrencies are the ISO-4217 codes Stripe treats as having no // minor unit (https://docs.stripe.com/currencies#zero-decimal). Amounts in // these currencies already arrive in the currency's base unit, not // hundredths, so they must not be divided by 100. Keys are lowercase to // match the lowercased lookup in formatCurrency. var zeroDecimalCurrencies = map[string]bool{ "bif": true, "clp": true, "djf": true, "gnf": true, "jpy": true, "kmf": true, "krw": true, "mga": true, "pyg": true, "rwf": true, "ugx": true, "vnd": true, "vuv": true, "xaf": true, "xof": true, "xpf": true, } // formatCurrency renders a minor-unit amount (cents, or whole units for // zero-decimal currencies) as a currency-aware display string, e.g. // "USD 10.00" or "JPY 1000". There is deliberately no hardcoded currency // symbol: Stripe supports 135+ currencies and a single glyph (e.g. "$") // would misrepresent every non-USD amount (finding #12) — the uppercase // ISO-4217 code is the only unambiguous prefix available without pulling in // locale-aware formatting. func formatCurrency(amount int32, currency string) string { code := strings.ToUpper(currency) if zeroDecimalCurrencies[strings.ToLower(currency)] { return fmt.Sprintf("%s %d", code, amount) } dollars := float64(amount) / 100 return fmt.Sprintf("%s %.2f", code, dollars) } // loadInvoicesData hydrates the invoices listing. func (h *OperatorPartialsHandler) loadInvoicesData(r *http.Request) InvoicesData { stripeConfigured, _ := configurationReadiness(h.IntegrationConfigs, "stripe") invoices, err := h.BillingQ.ListInvoices(r.Context()) if err != nil { h.Logger.Error("failed to list invoices", "error", err) return InvoicesData{Error: "Failed to load invoices"} } vms := make([]InvoiceViewModel, len(invoices)) for i, inv := range invoices { stripeInvoiceID := "" syncStatus := "not_mapped" if mapping, err := h.StripeQ.GetInvoiceMappingByInvoiceID(r.Context(), inv.InvoiceID); err == nil { if mapping.StripeInvoiceID.Valid { stripeInvoiceID = mapping.StripeInvoiceID.String } syncStatus = mapping.SyncStatus } dueDate := "" if inv.DueDate.Valid { dueDate = inv.DueDate.Time.Format("Jan 2, 2006") } paidAt := "" if inv.PaidAt.Valid { paidAt = inv.PaidAt.Time.Format("Jan 2, 2006") } vms[i] = InvoiceViewModel{ InvoiceID: inv.InvoiceID, BillingAccountID: inv.BillingAccountID, BillingAccountName: inv.BillingAccountName, Status: inv.Status, StatusClass: getInvoiceStatusBadgeClass(inv.Status, inv.AmountDue, inv.AmountPaid), AmountDue: formatCurrency(inv.AmountDue, inv.Currency), AmountPaid: formatCurrency(inv.AmountPaid, inv.Currency), Currency: inv.Currency, DueDate: dueDate, PaidAt: paidAt, StripeInvoiceID: stripeInvoiceID, StripeSyncStatus: syncStatus, CreatedAt: inv.CreatedAt.Format("Jan 2, 2006"), } } return InvoicesData{Invoices: vms, StripeConfigured: stripeConfigured} } // PaymentViewModel represents a payment for operator template rendering type PaymentViewModel struct { PaymentID string InvoiceID string BillingAccountID string BillingAccountName string Status string StatusClass string Amount string Currency string PaymentMethod string StripePaymentIntentID string StripeSyncStatus string FailedAt string CreatedAt string } // PaymentsData holds data for the payments list partial type PaymentsData struct { Payments []PaymentViewModel Error string // StripeConfigured gates the empty-view copy; see BillingAccountsData. StripeConfigured bool } // loadPaymentsData hydrates the payments listing. Each payment's real Stripe // sync status comes from a reverse-lookup on payment_mappings by local // payment_id (mappings are unique per payment_id); absent a mapping it stays // "not_mapped" (finding #14). func (h *OperatorPartialsHandler) loadPaymentsData(r *http.Request) PaymentsData { stripeConfigured, _ := configurationReadiness(h.IntegrationConfigs, "stripe") payments, err := h.BillingQ.ListPayments(r.Context()) if err != nil { h.Logger.Error("failed to list payments", "error", err) return PaymentsData{Error: "Failed to load payments"} } vms := make([]PaymentViewModel, len(payments)) for i, pay := range payments { stripePaymentIntentID := "" syncStatus := "not_mapped" if h.StripeQ != nil { if mapping, err := h.StripeQ.GetPaymentMappingByPaymentID(r.Context(), pay.PaymentID); err == nil { if mapping.StripePaymentIntentID.Valid { stripePaymentIntentID = mapping.StripePaymentIntentID.String } syncStatus = mapping.SyncStatus } } paymentMethod := "Unknown" if pay.PaymentMethodType.Valid { if pay.PaymentMethodType.String == "card" && pay.CardBrand.Valid && pay.CardLast4.Valid { paymentMethod = fmt.Sprintf("%s •••• %s", pay.CardBrand.String, pay.CardLast4.String) } else { paymentMethod = pay.PaymentMethodType.String } } failedAt := "" if pay.FailedAt.Valid { failedAt = pay.FailedAt.Time.Format("Jan 2, 2006") } vms[i] = PaymentViewModel{ PaymentID: pay.PaymentID, InvoiceID: pay.InvoiceID, BillingAccountID: pay.BillingAccountID, BillingAccountName: pay.BillingAccountName, Status: pay.Status, StatusClass: getPaymentStatusBadgeClass(pay.Status), Amount: formatCurrency(pay.Amount, pay.Currency), Currency: pay.Currency, PaymentMethod: paymentMethod, StripePaymentIntentID: stripePaymentIntentID, StripeSyncStatus: syncStatus, FailedAt: failedAt, CreatedAt: pay.CreatedAt.Format("Jan 2, 2006"), } } return PaymentsData{Payments: vms, StripeConfigured: stripeConfigured} } // PriceViewModel represents a price for operator template rendering type PriceViewModel struct { PriceID string ProductID string UnitAmount string Currency string RecurringInterval string IsRecurring bool TrialPeriodDays int32 IsActive bool // IsDefault marks the product's default price — the one price members are // offered (readiness, catalog, checkout all track it). At most one per // product, enforced by idx_prices_one_default_per_product. IsDefault bool StripePriceID string StripeSyncStatus string CreatedAt string } // ProductPricesData holds data for the product prices partial type ProductPricesData struct { ProductID string ProductName string Prices []PriceViewModel StripeDashboardURL string // StripeConfigured gates the per-row "Sync" affordance: syncing a price is // only actionable when the deployment has Stripe wired up. StripeConfigured bool FieldErrors web.FieldErrors Success string Error string } // maxStripeUnitAmount is Stripe's maximum unit amount, in the currency's // smallest unit (https://docs.stripe.com/api/prices/create — 99,999,999 = // $999,999.99). Enforced before the float→int32 conversion in CreatePrice so a // large amount can't silently overflow to a negative int32 that stores garbage // behind a success banner and then dead-letters at sync time (finding #47). const maxStripeUnitAmount = 99999999 // CreatePrice handles POST /partials/operator/products/{productID}/prices func (h *OperatorPartialsHandler) CreatePrice(w http.ResponseWriter, r *http.Request) { productID := r.PathValue("productID") if err := r.ParseForm(); err != nil { h.renderProductPricesPage(w, r, productID, "", "Invalid request") return } amountStr := strings.TrimSpace(r.FormValue("amount")) currency := strings.TrimSpace(r.FormValue("currency")) interval := r.FormValue("recurring_interval") trialDaysStr := strings.TrimSpace(r.FormValue("trial_period_days")) errs := web.New() var unitAmount int32 if amountStr == "" { errs.Set("amount", "Amount is required.") } else { dollars, parseErr := strconv.ParseFloat(amountStr, 64) switch { case parseErr != nil || math.IsNaN(dollars) || math.IsInf(dollars, 0) || dollars <= 0: // Rejects NaN/Inf (which slip past a bare `<= 0` check) and any // non-positive amount. errs.Set("amount", "Amount must be a positive number.") default: // Convert to integer cents, rejecting sub-cent precision (0.004 → // 0) and any value above Stripe's ceiling before the int32 cast, // so nothing can overflow to a negative unit_amount (finding #47). cents := dollars * 100 rounded := math.Round(cents) switch { case math.Abs(cents-rounded) > 0.001: errs.Set("amount", "Amount must be a whole number of cents (at most two decimal places).") case rounded > maxStripeUnitAmount: errs.Set("amount", "Amount is too large — the maximum is 999,999.99.") default: unitAmount = int32(rounded) } } } if currency == "" { errs.Set("currency", "Currency is required.") } recurringInterval := sql.NullString{} if interval != "" && interval != "one_time" { recurringInterval = sql.NullString{String: interval, Valid: true} } trialPeriodDays := sql.NullInt32{} if trialDaysStr != "" { days, parseErr := strconv.Atoi(trialDaysStr) if parseErr != nil || days < 0 { errs.Set("trial_period_days", "Trial period must be a non-negative integer.") } else if days > 0 { trialPeriodDays = sql.NullInt32{Int32: int32(days), Valid: true} } } if errs.Any() { h.renderProductPricesFormErrors(w, r, productID, errs) return } price, err := h.BillingQ.CreatePrice(r.Context(), billing.CreatePriceParams{ ProductID: productID, Currency: currency, UnitAmount: unitAmount, RecurringInterval: recurringInterval, TrialPeriodDays: trialPeriodDays, }) if err != nil { if fieldErrs, ok := web.FieldErrorsFromDB(err, nil); ok { h.renderProductPricesFormErrors(w, r, productID, fieldErrs) return } h.Logger.Error("failed to create price", slog.Any("error", err)) h.renderProductPricesPage(w, r, productID, "", "Failed to create price.") return } // Creating a price does NOT enqueue a Stripe sync: a create_stripe_price // entry with no synced product can only dead-letter. Stripe sync is the // explicit operator action SyncProductToStripe ("Sync to Stripe" on the // product readiness panel or per-row on the prices table). // Branch the success copy on Stripe readiness: when Stripe is unconfigured // the readiness panel shows a "not configured" alert with no "Sync to // Stripe" button, so pointing the operator at that affordance would lie // (finding #11). if price.IsDefault { // First price for the product — it became the default automatically. if h.StripeConfigured { h.renderProductPricesPage(w, r, productID, "Price created and set as this product's default. Use \"Sync to Stripe\" to make it purchasable.", "") } else { h.renderProductPricesPage(w, r, productID, "Price created and set as this product's default. Configure Stripe to enable payment.", "") } return } if h.StripeConfigured { h.renderProductPricesPage(w, r, productID, "Price created. Make it the default to offer it to members; sync it to Stripe to enable payment.", "") } else { h.renderProductPricesPage(w, r, productID, "Price created. Make it the default to offer it to members; configure Stripe to enable payment.", "") } } // MakeDefaultPrice handles POST /partials/operator/products/{productID}/prices/{priceID}/make-default. // It moves the product's default-price marker (the price members are offered) // to the named price: clear-then-set inside one transaction so the partial // unique index (one default per product) never sees two defaults. func (h *OperatorPartialsHandler) MakeDefaultPrice(w http.ResponseWriter, r *http.Request) { ctx := r.Context() productID := r.PathValue("productID") priceID := r.PathValue("priceID") price, err := h.BillingQ.GetPrice(ctx, priceID) if err != nil || price.ProductID != productID { h.renderProductPricesPage(w, r, productID, "", "Price not found for this product.") return } if price.IsDefault { h.renderProductPricesPage(w, r, productID, "This price is already the default.", "") return } if !price.IsActive { h.renderProductPricesPage(w, r, productID, "", "An inactive price cannot be made the default.") return } tx, err := h.Database.BeginTx(ctx, nil) if err != nil { h.Logger.Error("make-default: begin tx", slog.Any("error", err), slog.String("price_id", priceID)) h.renderProductPricesPage(w, r, productID, "", "Failed to update the default price.") return } defer tx.Rollback() //nolint:errcheck // no-op after Commit qtx := billing.New(tx) if err := qtx.ClearDefaultPrice(ctx, productID); err != nil { h.renderPriceWriteError(w, r, productID, "make-default: clear", priceID, err, "Failed to update the default price.") return } if _, err := qtx.MarkDefaultPrice(ctx, billing.MarkDefaultPriceParams{PriceID: priceID, ProductID: productID}); err != nil { h.renderPriceWriteError(w, r, productID, "make-default: mark", priceID, err, "Failed to update the default price.") return } if err := tx.Commit(); err != nil { h.renderPriceWriteError(w, r, productID, "make-default: commit", priceID, err, "Failed to update the default price.") return } h.renderProductPricesPage(w, r, productID, "Default price updated — members are now offered this price.", "") } // DeactivatePrice handles POST /partials/operator/products/{productID}/prices/{priceID}/deactivate. // Deactivating retires a price from the operator's offer surface (it stays // referenced by history: subscriptions, invoices). The default price cannot be // deactivated — members would lose the purchase path — so the operator must // move the default first; the sqlc query's is_default guard backstops this. func (h *OperatorPartialsHandler) DeactivatePrice(w http.ResponseWriter, r *http.Request) { ctx := r.Context() productID := r.PathValue("productID") priceID := r.PathValue("priceID") price, err := h.BillingQ.GetPrice(ctx, priceID) if err != nil || price.ProductID != productID { h.renderProductPricesPage(w, r, productID, "", "Price not found for this product.") return } if price.IsDefault { h.renderProductPricesPage(w, r, productID, "", "The default price cannot be deactivated — make another price the default first.") return } if !price.IsActive { h.renderProductPricesPage(w, r, productID, "This price is already inactive.", "") return } if _, err := h.BillingQ.DeactivatePrice(ctx, priceID); err != nil { if errors.Is(err, sql.ErrNoRows) { // The is_default guard in the query rejected the write (the price // became the default between our read and the update). h.renderProductPricesPage(w, r, productID, "", "The default price cannot be deactivated — make another price the default first.") return } h.renderPriceWriteError(w, r, productID, "deactivate price", priceID, err, "Failed to deactivate the price.") return } h.renderProductPricesPage(w, r, productID, "Price deactivated — it is no longer offered.", "") } // renderPriceWriteError routes a failed price write to the §6 error contract: // recognizable constraint violations render as 422 + FieldErrors on the prices // view (web.FieldErrorsFromDB), anything else is logged and rendered as a // generic banner — never err.Error(). func (h *OperatorPartialsHandler) renderPriceWriteError(w http.ResponseWriter, r *http.Request, productID, op, priceID string, err error, generic string) { if fieldErrs, ok := web.FieldErrorsFromDB(err, nil); ok { h.renderProductPricesFormErrors(w, r, productID, fieldErrs) return } h.Logger.Error(op, slog.Any("error", err), slog.String("product_id", productID), slog.String("price_id", priceID)) h.renderProductPricesPage(w, r, productID, "", generic) } // SyncProductToStripe handles POST /partials/operator/products/{productID}/sync-stripe. // It makes the "Payment processing" purchasability precondition actionable: it // drives the product and one of its active prices to Stripe-mapped state by // writing the mapping rows as pending and enqueuing the catalog sync. The price // is selected by an optional price_id form value (the per-row "Sync" button on // the prices table); absent that it is the product's default price (the // readiness-panel button) — the same price readiness and checkout track. // // Product-idempotent: when the product mapping is already synced, only a // create_stripe_price entry is enqueued for the selected price — re-syncing to // enable an additional price must never create a duplicate Stripe product. // Per-price guards — no-op when Stripe is unconfigured, when there is no active // price, or when the selected price's sync is already pending/synced, so repeat // clicks never create duplicate Stripe objects. func (h *OperatorPartialsHandler) SyncProductToStripe(w http.ResponseWriter, r *http.Request) { ctx := r.Context() productID := r.PathValue("productID") if !h.StripeConfigured { h.renderProductEditPage(w, r, productID, "", "Stripe is not configured for this deployment, so pricing cannot be synced.") return } product, err := h.BillingQ.GetProductByID(ctx, productID) if err != nil { h.Logger.Error("sync-stripe: get product", slog.Any("error", err), slog.String("product_id", productID)) h.renderProductEditPage(w, r, productID, "", "Failed to load the product.") return } prices, err := h.BillingQ.ListPricesByProduct(ctx, productID) if err != nil || len(prices) == 0 { h.renderProductEditPage(w, r, productID, "", "Add an active price before syncing to Stripe.") return } // Select the price to sync: explicit price_id first, else the default // price, else (defensively, when no default is marked) the oldest active. price := prices[0] for _, p := range prices { if p.IsDefault { price = p break } } if requested := strings.TrimSpace(r.FormValue("price_id")); requested != "" { found := false for _, p := range prices { if p.PriceID == requested { price, found = p, true break } } if !found { h.renderProductEditPage(w, r, productID, "", "That price is not an active price of this product, so it cannot be synced.") return } } // Idempotency + retry. The outbox executors create a new Stripe object per run, // so never enqueue a duplicate for an already-synced or in-flight sync. But a // terminally-failed (dead-lettered) sync leaves the mapping stuck at 'pending', // so treat that as a retry — re-drive the dead-lettered entries — rather than a // no-op on the stuck pending mapping. m, mErr := h.StripeQ.GetPriceMappingByPriceID(ctx, price.PriceID) if mErr == nil && m.StripePriceID.Valid { h.renderProductEditPage(w, r, productID, "This price is already synced to Stripe.", "") return } if failed, _ := h.stripeSyncFailure(ctx, productID, price.PriceID); failed { if _, err := h.Database.ExecContext(ctx, `UPDATE core.outbox SET status = 'pending', attempts = 0, next_attempt_at = NOW(), error_message = NULL, updated_at = NOW() WHERE provider = 'stripe' AND status = 'dead_letter' AND ( (action_type = 'create_stripe_product' AND payload->>'product_id' = $1) OR (action_type = 'create_stripe_price' AND payload->>'price_id' = $2) )`, productID, price.PriceID); err != nil { h.Logger.Error("sync-stripe: reset dead-lettered outbox", slog.Any("error", err), slog.String("product_id", productID)) h.renderProductEditPage(w, r, productID, "", "Failed to retry the sync.") return } h.renderProductEditPage(w, r, productID, "Retrying sync to Stripe — the mapping will appear here shortly.", "") return } if mErr == nil && m.SyncStatus == "pending" { h.renderProductEditPage(w, r, productID, "A Stripe sync is already in progress for this price.", "") return } // Serialize concurrent syncs of this product and make the mapping writes // and the outbox enqueue atomic (finding #23). Two failure modes this // closes: (1) a mid-flight failure between the mapping upserts and the // outbox INSERT used to strand the price at sync_status='pending' with no // outbox row — permanently "sync in progress" and unretryable, since the // dead-letter retry path above only matches outbox rows that were never // created; (2) two concurrent Sync clicks both cleared the check-then-act // guards above and enqueued duplicate create_stripe_product entries. A // per-product advisory xact lock (same idiom as fulfillment/reconcile.go) // serializes them, and re-reading the mappings under the lock makes the // loser a no-op instead of a duplicate. tx, err := h.Database.BeginTx(ctx, nil) if err != nil { h.Logger.Error("sync-stripe: begin tx", slog.Any("error", err), slog.String("product_id", productID)) h.renderProductEditPage(w, r, productID, "", "Failed to start the sync.") return } defer tx.Rollback() //nolint:errcheck // no-op after Commit if _, err := tx.ExecContext(ctx, `SELECT pg_advisory_xact_lock(hashtext($1)::bigint)`, productID); err != nil { h.Logger.Error("sync-stripe: advisory lock", slog.Any("error", err), slog.String("product_id", productID)) h.renderProductEditPage(w, r, productID, "", "Failed to start the sync.") return } qtx := stripedb.New(tx) // Re-check the price mapping under the lock: a concurrent sync that ran // between our pre-lock guards and acquiring the lock may already have // driven this price to synced or pending. if pm, err := qtx.GetPriceMappingByPriceID(ctx, price.PriceID); err == nil { if pm.StripePriceID.Valid { h.renderProductEditPage(w, r, productID, "This price is already synced to Stripe.", "") return } if pm.SyncStatus == "pending" { h.renderProductEditPage(w, r, productID, "A Stripe sync is already in progress for this price.", "") return } } // Product idempotency: a mapping with a live stripe_product_id means the // Stripe product already exists — enqueue ONLY the price sync below, never // a second create_stripe_product (which would create a duplicate Stripe // product). A pending product mapping with no ID yet means the product // create is still in flight; adding a price entry now could land before // its parent product, so treat it as in-progress. productSynced := false if pm, err := qtx.GetProductMappingByProductID(ctx, productID); err == nil { if pm.StripeProductID.Valid { productSynced = true } else if pm.SyncStatus == "pending" { h.renderProductEditPage(w, r, productID, "A Stripe sync is already in progress for this product.", "") return } } // Write pending mappings so the readiness panel immediately shows "sync // pending"; the executors upsert them to "synced" when the sync lands. An // already-synced product mapping is left untouched. if !productSynced { if _, err := qtx.UpsertProductMapping(ctx, stripedb.UpsertProductMappingParams{ ProductID: productID, SyncStatus: "pending", }); err != nil { h.Logger.Error("sync-stripe: upsert product mapping", slog.Any("error", err), slog.String("product_id", productID)) h.renderProductEditPage(w, r, productID, "", "Failed to start the sync.") return } } if _, err := qtx.UpsertPriceMapping(ctx, stripedb.UpsertPriceMappingParams{ PriceID: price.PriceID, SyncStatus: "pending", }); err != nil { h.Logger.Error("sync-stripe: upsert price mapping", slog.Any("error", err), slog.String("price_id", price.PriceID)) h.renderProductEditPage(w, r, productID, "", "Failed to start the sync.") return } pricePayload := map[string]any{ "price_id": price.PriceID, "product_id": productID, "unit_amount": price.UnitAmount, "currency": price.Currency, "recurring_interval": price.RecurringInterval.String, } if productSynced { if err := integration.Enqueue(ctx, tx, "stripe", "create_stripe_price", pricePayload); err != nil { h.Logger.Error("sync-stripe: enqueue outbox", slog.Any("error", err), slog.String("product_id", productID)) h.renderProductEditPage(w, r, productID, "", "Failed to enqueue the sync.") return } if err := tx.Commit(); err != nil { h.Logger.Error("sync-stripe: commit", slog.Any("error", err), slog.String("product_id", productID)) h.renderProductEditPage(w, r, productID, "", "Failed to enqueue the sync.") return } h.renderProductEditPage(w, r, productID, "Price sync to Stripe enqueued — the mapping will appear here shortly.", "") return } productPayload := map[string]any{ "product_id": productID, "name": product.Name, "description": product.Description.String, "display_category": product.DisplayCategory.String, } // Two Enqueue calls in the same tx rather than a batch API (task 1.7): // atomicity comes from the shared transaction, not from the helper. if err := integration.Enqueue(ctx, tx, "stripe", "create_stripe_product", productPayload); err != nil { h.Logger.Error("sync-stripe: enqueue outbox", slog.Any("error", err), slog.String("product_id", productID)) h.renderProductEditPage(w, r, productID, "", "Failed to enqueue the sync.") return } if err := integration.Enqueue(ctx, tx, "stripe", "create_stripe_price", pricePayload); err != nil { h.Logger.Error("sync-stripe: enqueue outbox", slog.Any("error", err), slog.String("product_id", productID)) h.renderProductEditPage(w, r, productID, "", "Failed to enqueue the sync.") return } if err := tx.Commit(); err != nil { h.Logger.Error("sync-stripe: commit", slog.Any("error", err), slog.String("product_id", productID)) h.renderProductEditPage(w, r, productID, "", "Failed to enqueue the sync.") return } h.renderProductEditPage(w, r, productID, "Sync to Stripe enqueued — the mapping will appear here shortly.", "") } // stripeSyncFailure reports whether the product's Stripe catalog-sync has // terminally failed (a create_stripe_product / create_stripe_price outbox entry // reached dead_letter), returning the recorded error. Transient 'failed' rows // (still auto-retried by the outbox poller) are treated as in-flight, not failed. // priceID may be empty (no active price yet) — then only the product action matches. func (h *OperatorPartialsHandler) stripeSyncFailure(ctx context.Context, productID, priceID string) (bool, string) { var errMsg string err := h.Database.QueryRowContext(ctx, `SELECT COALESCE(error_message, '') FROM core.outbox WHERE provider = 'stripe' AND status = 'dead_letter' AND ( (action_type = 'create_stripe_product' AND payload->>'product_id' = $1) OR (action_type = 'create_stripe_price' AND payload->>'price_id' = $2) ) ORDER BY updated_at DESC LIMIT 1`, productID, priceID).Scan(&errMsg) if err != nil { return false, "" } return true, errMsg } // renderProductPricesFormErrors re-renders the prices page (with the inline // add-price form) with 422 + FieldErrors populated. Routes through // renderProductPricesPage's full data load by setting FieldErrors after the // load returns — safe because no body write happens before we set the header. func (h *OperatorPartialsHandler) renderProductPricesFormErrors(w http.ResponseWriter, r *http.Request, productID string, errs web.FieldErrors) { // Re-render the full composite so the failing add-price form appears // alongside the edit form and readiness panel it now shares a page with. w.WriteHeader(http.StatusUnprocessableEntity) edit, ok := h.loadProductEditData(r, productID) if !ok { h.renderProductsPage(w, r, "", "Product not found") return } prices := h.loadProductPricesData(r, productID) prices.FieldErrors = errs h.Templates.Render(w, "operator_product_detail.html", ProductDetailData{ Edit: edit, Prices: prices, }) } // renderProductPricesPage re-renders the product composite in place. Retained as // a thin alias so the price mutation call sites (CreatePrice) keep working; the // prices view now lives on the composite detail page. func (h *OperatorPartialsHandler) renderProductPricesPage(w http.ResponseWriter, r *http.Request, productID string, success string, errMsg string) { h.renderProductDetailBody(w, r, productID, success, errMsg) } // loadProductPricesData hydrates the prices view model for one product: the // price rows, each with its Stripe mapping status, plus the dashboard link. // Best-effort — a query failure yields a partial result with Error set rather // than aborting, since this feeds the composite detail page where the product's // existence is already established. Guards a nil StripeQ (Stripe unconfigured). func (h *OperatorPartialsHandler) loadProductPricesData(r *http.Request, productID string) ProductPricesData { data := ProductPricesData{ ProductID: productID, StripeDashboardURL: h.StripeDashboardURL, StripeConfigured: h.StripeConfigured, } if product, err := h.BillingQ.GetProductByID(r.Context(), productID); err == nil { data.ProductName = product.Name } prices, err := h.BillingQ.ListPricesByProduct(r.Context(), productID) if err != nil { h.Logger.Error("failed to list prices", "error", err, "product_id", productID) data.Error = "Failed to load prices" return data } vms := make([]PriceViewModel, len(prices)) for i, price := range prices { stripePriceID := "" syncStatus := "not_mapped" if h.StripeQ != nil { if mapping, err := h.StripeQ.GetPriceMappingByPriceID(r.Context(), price.PriceID); err == nil { if mapping.StripePriceID.Valid { stripePriceID = mapping.StripePriceID.String } syncStatus = mapping.SyncStatus } } isRecurring := price.RecurringInterval.Valid interval := "" if isRecurring { interval = price.RecurringInterval.String } vms[i] = PriceViewModel{ PriceID: price.PriceID, ProductID: price.ProductID, UnitAmount: formatCurrency(price.UnitAmount, price.Currency), Currency: price.Currency, RecurringInterval: interval, IsRecurring: isRecurring, TrialPeriodDays: price.TrialPeriodDays.Int32, IsActive: price.IsActive, IsDefault: price.IsDefault, StripePriceID: stripePriceID, StripeSyncStatus: syncStatus, CreatedAt: price.CreatedAt.Format("Jan 2, 2006"), } } data.Prices = vms return data } // getStatusBadgeClass returns the Bootstrap badge class for a subscription status func getStatusBadgeClass(status string) string { switch status { case "active", "trialing": return "success" case "past_due", "unpaid", "incomplete": return "danger" case "canceled", "ended", "incomplete_expired", "paused": return "secondary" default: return "secondary" } } // getInvoiceStatusBadgeClass returns the Bootstrap badge class for an invoice status func getInvoiceStatusBadgeClass(status string, amountDue, amountPaid int32) string { switch status { case "paid": if amountDue == amountPaid { return "success" } return "warning" case "open", "draft": return "warning" case "void", "uncollectible": return "secondary" default: return "secondary" } } // getPaymentStatusBadgeClass returns the Bootstrap badge class for a payment status func getPaymentStatusBadgeClass(status string) string { switch status { case "succeeded": return "success" case "pending": return "warning" case "failed": return "danger" case "canceled": return "secondary" default: return "secondary" } } // getSyncStatusBadgeClass returns the Bootstrap badge class for a sync status func getSyncStatusBadgeClass(status string) string { switch status { case "synced": return "success" case "pending": return "warning" case "failed", "dead_letter": return "danger" case "deleted": return "secondary" default: return "secondary" } } // GetStripeEntityURL returns the Stripe dashboard deep-link for an entity. func (h *OperatorPartialsHandler) GetStripeEntityURL(entityType, stripeID string) string { return stripeEntityURL(h.StripeDashboardURL, entityType, stripeID) } // stripeEntityURL builds a Stripe dashboard deep-link for an entity, or "" when // the dashboard base URL is unconfigured or the ID is empty — callers then fall // back to rendering the raw ID as plain text. entityType is a Stripe dashboard // path segment such as "customers" or "prices". Registered in the operator // template FuncMap as `stripeEntityURL` so templates can wrap Stripe IDs in // links gated on the dashboard URL being set (finding #39). func stripeEntityURL(dashboardURL, entityType, stripeID string) string { if dashboardURL == "" || stripeID == "" { return "" } return fmt.Sprintf("%s/%s/%s", dashboardURL, entityType, stripeID) } // Helper to get current time for templates func now() time.Time { return time.Now() }