amadan

carlos / platform Public

Add the team quantity worker behind an explicit proration policy

Member additions on a live team subscription were recorded and never
consumed, so one added member held that account's monthly freeze for
good. The worker raises the subscription item to the peak desired
quantity of the unprocessed additions (removals apply at renewal, so a
seat added and removed inside a period is still paid for), records the
result through the ordinary reconcile, and only then marks the additions.
It reads the provider first, so a lost response is repaired without a
second update, and a provider quantity it neither billed nor wants is
filed for review.

When the proration is collected is Paul's decision:
CONSOLE_BILLING_PRORATION is immediate (always_invoice) or next_invoice
(create_prorations). Unset, the worker makes no provider call and marks
nothing, and the hold stays.

Not done: lowering the quantity at renewal for members who left. Verified
with a fake provider (seven QuantityWorker tests); no real Stripe.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude 1fde082a85445f6be4626998b0e8e346547c612f parent b448ee5
6 files changed, +631 −1
  • cmd/console/billing.go +6 −0
  • internal/billingstripe/client.go +21 −1
  • internal/console/billingpass.go +7 −0
  • internal/console/billingquantity.go +281 −0
  • internal/console/billingquantity_test.go +307 −0
  • internal/console/billingservice.go +9 −0
diff --git a/cmd/console/billing.go b/cmd/console/billing.go
index de4227d55..12a633a47 100644
--- a/cmd/console/billing.go
+++ b/cmd/console/billing.go
@@ -26,6 +26,12 @@ func billingConfig(getenv func(string) string) (*console.BillingConfig, error) {
PortalConfiguration: getenv("CONSOLE_STRIPE_PORTAL_CONFIGURATION"),
HouseCoupon: getenv("CONSOLE_STRIPE_HOUSE_COUPON"),
CountryBands: plans.CountryBands(),
+ // The proration collection policy is a product decision with no
+ // default: unset leaves member additions pending.
+ Proration: getenv("CONSOLE_BILLING_PRORATION"),
+ }
+ if cfg.Proration != "" && cfg.Proration != "immediate" && cfg.Proration != "next_invoice" {
+ return nil, errBillingConfiguration
}
prefix := "rk_test_"
if cfg.LiveMode {
diff --git a/internal/billingstripe/client.go b/internal/billingstripe/client.go
index 93657ba79..9f4620e90 100644
--- a/internal/billingstripe/client.go
+++ b/internal/billingstripe/client.go
@@ -603,19 +603,39 @@ func (c *Client) GetSubscription(ctx context.Context, id string) (*Subscription,
return &sub, nil
}
+// Proration behaviours UpdateQuantityWith accepts. They are Stripe's own
+// values for proration_behavior.
+const (
+ ProrationAlwaysInvoice = "always_invoice" // invoice and collect the proration now
+ ProrationCreateProrations = "create_prorations" // add it to the next invoice
+ ProrationNone = "none" // change the quantity, prorate nothing
+)
+
// UpdateQuantity changes one subscription item's quantity, invoicing the
// proration immediately (always_invoice) rather than deferring it to the
// next regular invoice.
func (c *Client) UpdateQuantity(ctx context.Context, itemID string, quantity int64, idempotencyKey string) (*SubscriptionItem, error) {
+ return c.UpdateQuantityWith(ctx, itemID, quantity, ProrationAlwaysInvoice, idempotencyKey)
+}
+
+// UpdateQuantityWith changes one subscription item's quantity under the
+// named proration behaviour. When the proration is collected is a product
+// decision the caller carries; nothing here defaults it.
+func (c *Client) UpdateQuantityWith(ctx context.Context, itemID string, quantity int64, proration, idempotencyKey string) (*SubscriptionItem, error) {
if !validID(itemID) {
return nil, &Error{Code: CodeInvalidID}
}
if quantity < 1 {
return nil, &Error{Code: CodeInvalidParams}
}
+ switch proration {
+ case ProrationAlwaysInvoice, ProrationCreateProrations, ProrationNone:
+ default:
+ return nil, &Error{Code: CodeInvalidParams}
+ }
form := url.Values{}
form.Set("quantity", strconv.FormatInt(quantity, 10))
- form.Set("proration_behavior", "always_invoice")
+ form.Set("proration_behavior", proration)
var item SubscriptionItem
if err := c.do(ctx, http.MethodPost, "/v1/subscription_items/"+itemID, form, idempotencyKey, &item); err != nil {
return nil, err
diff --git a/internal/console/billingpass.go b/internal/console/billingpass.go
index 13806f1e0..ae4e4ff93 100644
--- a/internal/console/billingpass.go
+++ b/internal/console/billingpass.go
@@ -36,6 +36,13 @@ func (s *Server) StartBillingPass(ctx context.Context) {
log.Printf("console: billing reconciliation: %s", billingDiagnostic(err))
}
cancel()
+ // Member additions next (billingquantity.go): they change the
+ // subscription quantity, which is history the invoice pass reads.
+ quantities, cancelQuantities := context.WithTimeout(ctx, 2*time.Minute)
+ if err := s.billingQuantityPass(quantities, time.Now().UTC()); err != nil {
+ log.Printf("console: billing quantities: %s", billingDiagnostic(err))
+ }
+ cancelQuantities()
// Subscription state first, invoices second: a renewal or
// cancellation learned just now is history the invoice pass reads.
invoicing, cancelInvoicing := context.WithTimeout(ctx, billingInvoicePassTimeout)
diff --git a/internal/console/billingquantity.go b/internal/console/billingquantity.go
new file mode 100644
index 000000000..f0200eed1
--- /dev/null
+++ b/internal/console/billingquantity.go
@@ -0,0 +1,281 @@
+package console
+
+import (
+ "context"
+ "database/sql"
+ "errors"
+ "fmt"
+ "strconv"
+ "time"
+
+ "github.com/carlosframework/platform/internal/billingstripe"
+)
+
+// The team quantity worker (pricing spec, "Member changes update the
+// subscription quantity. Additions prorate; people who leave come off at
+// renewal").
+//
+// billing_quantity_events records every member added to or removed from an
+// account with a live team subscription. Until this file nothing consumed
+// them, so one added member held that account's monthly freeze for good.
+// The worker sends the quantity the additions call for to the provider,
+// records the result through the ordinary reconcile path, and only then
+// marks the additions processed.
+//
+// When the prorated amount is collected is a product decision
+// (BillingConfig.Proration). With no policy configured the worker does
+// nothing at all: no provider call, no mark, and the hold stays.
+//
+// What quantity: removals apply at renewal, so inside a period the billed
+// quantity only ever rises. The target is the PEAK desired quantity over
+// the unprocessed additions, not today's member count. A member added on
+// Monday and removed on Tuesday still occupied a seat that is paid for
+// until renewal.
+//
+// Order, and why it is crash safe:
+//
+// 1. Read the subscription from the provider and check it is the one the
+// console bills. If it already carries the target, an earlier attempt
+// landed and nothing is sent.
+// 2. Otherwise update the item to the target, under an idempotency key
+// fixed by the account, subscription, newest addition and target.
+// 3. Reconcile the subscription: the authoritative read that moves
+// billing_accounts.quantity and appends the assignment revision.
+// 4. Mark the additions processed, in a transaction that re-checks the
+// billed quantity covers them.
+//
+// A crash after 2 is repaired by 1 on the next pass; after 3, by 4 alone.
+// A provider quantity that is neither what the console bills nor the target
+// is filed for review and nothing is sent.
+//
+// Not done here: lowering the quantity at renewal for members who left.
+// Until that exists a renewal bills the old quantity, which must be fixed
+// before team plans are sold (see AGENT_HANDOFF.md).
+
+const (
+ billingProrationImmediate = "immediate"
+ billingProrationNextInvoice = "next_invoice"
+
+ quantityOutcomeCharged = "charged"
+ quantityOutcomeIncluded = "included"
+
+ billingQuantityMismatchReason = "billing_quantity_mismatch"
+)
+
+// prorationBehavior maps the configured policy to the provider's behaviour
+// and to the outcome recorded on the events. ok is false when no policy is
+// configured.
+func (cfg BillingConfig) prorationBehavior() (behavior, outcome string, ok bool) {
+ switch cfg.Proration {
+ case billingProrationImmediate:
+ return billingstripe.ProrationAlwaysInvoice, quantityOutcomeCharged, true
+ case billingProrationNextInvoice:
+ return billingstripe.ProrationCreateProrations, quantityOutcomeIncluded, true
+ }
+ return "", "", false
+}
+
+// pendingQuantityAccounts lists accounts with an unprocessed addition bound
+// to a subscription, in account order.
+func (s *Store) pendingQuantityAccounts() ([]int64, error) {
+ rows, err := s.db.Query(`SELECT DISTINCT account_id FROM billing_quantity_events WHERE change=? AND processed_at=0 AND subscription<>'' ORDER BY account_id`, quantityEventAddition)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+ var out []int64
+ for rows.Next() {
+ var id int64
+ if err := rows.Scan(&id); err != nil {
+ return nil, err
+ }
+ out = append(out, id)
+ }
+ return out, rows.Err()
+}
+
+// pendingQuantityAdditions summarises the unprocessed additions recorded
+// against subscription: how many, the newest generation, and the peak
+// desired quantity among them.
+func pendingQuantityAdditions(q interface {
+ QueryRow(string, ...any) *sql.Row
+}, accountID int64, subscription string, maxGeneration int64) (count, newest, peak int64, err error) {
+ err = q.QueryRow(`SELECT COUNT(*), COALESCE(MAX(generation),0), COALESCE(MAX(desired_quantity),0) FROM billing_quantity_events
+ WHERE account_id=? AND change=? AND processed_at=0 AND subscription=? AND (?=0 OR generation<=?)`,
+ accountID, quantityEventAddition, subscription, maxGeneration, maxGeneration).Scan(&count, &newest, &peak)
+ return count, newest, peak, err
+}
+
+// markQuantityAdditionsProcessed marks the unprocessed additions of
+// subscription up to maxGeneration, once the billed quantity covers them.
+// It refuses with errBillingPending when the reconcile that moves the
+// billed quantity has not landed.
+func (s *Store) markQuantityAdditionsProcessed(accountID int64, subscription string, maxGeneration int64, outcome, reference string, now time.Time) error {
+ tx, err := s.db.Begin()
+ if err != nil {
+ return err
+ }
+ defer tx.Rollback()
+ // Take the membership write lock first: no addition can be recorded
+ // between the check and the mark.
+ if _, err := lockMembershipWritesTx(tx); err != nil {
+ return err
+ }
+ count, _, peak, err := pendingQuantityAdditions(tx, accountID, subscription, maxGeneration)
+ if err != nil {
+ return err
+ }
+ if count == 0 {
+ return tx.Commit()
+ }
+ var billed int64
+ var current string
+ if err := tx.QueryRow(`SELECT quantity,subscription FROM billing_accounts WHERE account_id=?`, accountID).Scan(&billed, &current); err != nil {
+ return err
+ }
+ if current != subscription || billed < peak {
+ return errBillingPending
+ }
+ if _, err := tx.Exec(`UPDATE billing_quantity_events SET processed_at=max(?,occurred_at_ns/1000000000+1), processing_outcome=?, processing_reference=?
+ WHERE account_id=? AND change=? AND processed_at=0 AND subscription=? AND generation<=?`,
+ now.Unix(), outcome, reference, accountID, quantityEventAddition, subscription, maxGeneration); err != nil {
+ return err
+ }
+ if err := refreshQuantityChangeTx(tx, accountID, now); err != nil {
+ return err
+ }
+ return tx.Commit()
+}
+
+// billingQuantityPass processes every account with pending additions. With
+// no proration policy configured it returns at once.
+func (s *Server) billingQuantityPass(ctx context.Context, now time.Time) error {
+ b := s.billing
+ if b == nil {
+ return nil
+ }
+ behavior, outcome, ok := b.cfg.prorationBehavior()
+ if !ok {
+ return nil
+ }
+ ids, err := s.store.pendingQuantityAccounts()
+ if err != nil {
+ return err
+ }
+ var failures []error
+ for _, id := range ids {
+ if ctx.Err() != nil {
+ return errors.Join(append(failures, ctx.Err())...)
+ }
+ if err := s.processQuantityAdditions(ctx, id, behavior, outcome, now); err != nil {
+ failures = append(failures, err)
+ }
+ }
+ return errors.Join(failures...)
+}
+
+func (s *Server) processQuantityAdditions(ctx context.Context, accountID int64, behavior, outcome string, now time.Time) error {
+ acct, err := s.store.AccountByID(accountID)
+ if err != nil {
+ return err
+ }
+ if acct == nil {
+ return nil
+ }
+ state, newest, reference, done, err := s.raiseQuantity(ctx, acct, behavior, now)
+ if err != nil || done {
+ return err
+ }
+ if reference != "" {
+ // The provider carries the target. The reconcile is the read that
+ // moves billing_accounts.quantity and records the assignment.
+ if err := s.reconcileBillingSubscription(ctx, state.Subscription, "", now); err != nil {
+ return err
+ }
+ } else {
+ // The quantity already billed covers every addition (a team below
+ // its minimum, or a repair after a crash): nothing was sent now.
+ outcome = quantityOutcomeIncluded
+ reference = "quantity_already_billed:" + strconv.FormatInt(state.Quantity, 10)
+ }
+ return s.store.markQuantityAdditionsProcessed(accountID, state.Subscription, newest, outcome, reference, now)
+}
+
+// raiseQuantity is steps 1 and 2 under the account's billing lock. done is
+// true when there is nothing to process or the subscription is not in a
+// state to be raised; reference is non-empty when the provider carries the
+// target and a reconcile must follow.
+func (s *Server) raiseQuantity(ctx context.Context, acct *Account, behavior string, now time.Time) (state billingState, newest int64, reference string, done bool, err error) {
+ b := s.billing
+ defer b.locks.lock(acct.Sqid, "billing")()
+ state, err = s.store.billingAccount(acct.ID)
+ if err != nil {
+ return state, 0, "", false, err
+ }
+ if state.Subscription == "" {
+ return state, 0, "", true, nil
+ }
+ count, newest, peak, err := pendingQuantityAdditions(s.store.db, acct.ID, state.Subscription, 0)
+ if err != nil {
+ return state, 0, "", false, err
+ }
+ if count == 0 {
+ return state, 0, "", true, nil
+ }
+ if peak <= state.Quantity {
+ return state, newest, "", false, nil
+ }
+ // Only a healthy subscription is raised: adding a charge to one whose
+ // last payment failed, or that is over, helps nobody. The additions
+ // stay pending and hold.
+ if state.Status != "active" {
+ return state, 0, "", true, nil
+ }
+ sub, err := b.client.GetSubscription(ctx, state.Subscription)
+ if err != nil {
+ return state, 0, "", false, err
+ }
+ if sub.ID != state.Subscription || sub.LiveMode != b.cfg.LiveMode || sub.CustomerID != state.Customer ||
+ sub.Metadata["account_sqid"] != acct.Sqid || len(sub.Items) != 1 || sub.ItemsHasMore ||
+ sub.Items[0].PriceID != b.cfg.Prices[billingPriceKey(state.Terms, state.Plan, state.Band)] {
+ return state, 0, "", false, errBillingIdentity
+ }
+ if sub.Status != "active" {
+ return state, 0, "", true, nil
+ }
+ item := sub.Items[0]
+ switch item.Quantity {
+ case peak:
+ // An earlier attempt landed and was not recorded.
+ case state.Quantity:
+ key := fmt.Sprintf("carloku-qty-%s-%s-%d-%d", acct.Sqid, state.Subscription, newest, peak)
+ updated, err := b.client.UpdateQuantityWith(ctx, item.ID, peak, behavior, key)
+ if err != nil {
+ return state, 0, "", false, err
+ }
+ if updated.ID != item.ID || updated.Quantity != peak {
+ return state, 0, "", false, errBillingIdentity
+ }
+ default:
+ // Neither what the console bills nor what it wants: someone else
+ // changed the subscription. A person decides.
+ if ferr := s.store.fileBillingReview(acct.ID, state.Subscription, billingQuantityMismatchReason, now); ferr != nil {
+ return state, 0, "", false, ferr
+ }
+ return state, 0, "", true, nil
+ }
+ return state, newest, item.ID + ":quantity=" + strconv.FormatInt(peak, 10) + ":" + behavior, false, nil
+}
+
+// fileBillingReview files one review occurrence in its own transaction.
+func (s *Store) fileBillingReview(accountID int64, subscription, reason string, observed time.Time) error {
+ tx, err := s.db.Begin()
+ if err != nil {
+ return err
+ }
+ defer tx.Rollback()
+ if err := fileBillingReviewTx(tx, accountID, subscription, reason, observed); err != nil {
+ return err
+ }
+ return tx.Commit()
+}
diff --git a/internal/console/billingquantity_test.go b/internal/console/billingquantity_test.go
new file mode 100644
index 000000000..8e15238fb
--- /dev/null
+++ b/internal/console/billingquantity_test.go
@@ -0,0 +1,307 @@
+package console
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "net/http"
+ "net/http/httptest"
+ "strconv"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/billingstripe"
+ "github.com/carlosframework/platform/internal/plans"
+)
+
+// quantityStripe is a fake provider holding one team subscription whose
+// quantity the worker may raise.
+type quantityStripe struct {
+ mu sync.Mutex
+ sqid string
+ quantity int64
+ status string
+ gets int
+ posts []map[string]string // quantity, proration_behavior, key
+ failPost bool
+ start time.Time
+ failAfter bool // accept the POST, then answer 500: a lost response
+}
+
+func (q *quantityStripe) serve(w http.ResponseWriter, r *http.Request) {
+ q.mu.Lock()
+ defer q.mu.Unlock()
+ w.Header().Set("Content-Type", "application/json")
+ item := func() map[string]any {
+ return map[string]any{"id": "si_team", "price": map[string]string{"id": "price_community_A"}, "quantity": q.quantity}
+ }
+ switch {
+ case r.URL.Path == "/v1/subscriptions/sub_team" && r.Method == http.MethodGet:
+ q.gets++
+ _ = json.NewEncoder(w).Encode(map[string]any{"id": "sub_team", "customer": "cus_team", "status": q.status, "livemode": true,
+ "created": q.start.Unix(), "current_period_start": q.start.Unix(), "current_period_end": q.start.AddDate(1, 0, 0).Unix(),
+ "metadata": map[string]string{"account_sqid": q.sqid, "plan": plans.PlanCommunity, "terms_version": plans.CurrentTermsVersion, "band": "A", "checkout_intent": ""},
+ "items": map[string]any{"data": []any{item()}, "has_more": false}})
+ case r.URL.Path == "/v1/subscription_items/si_team" && r.Method == http.MethodPost:
+ _ = r.ParseForm()
+ if q.failPost {
+ w.WriteHeader(http.StatusInternalServerError)
+ return
+ }
+ n, _ := strconv.ParseInt(r.PostForm.Get("quantity"), 10, 64)
+ q.posts = append(q.posts, map[string]string{"quantity": r.PostForm.Get("quantity"), "proration_behavior": r.PostForm.Get("proration_behavior"), "key": r.Header.Get("Idempotency-Key")})
+ q.quantity = n
+ if q.failAfter {
+ q.failAfter = false
+ w.WriteHeader(http.StatusInternalServerError)
+ return
+ }
+ _ = json.NewEncoder(w).Encode(item())
+ default:
+ w.WriteHeader(http.StatusNotFound)
+ }
+}
+
+// newQuantityFixture is a live Community account at its floor of ten
+// members, billed for ten, with its entitlement published.
+func newQuantityFixture(t *testing.T, proration string) (*consoleFixture, *quantityStripe, time.Time) {
+ t.Helper()
+ fx := newConsoleFixture(t)
+ now := time.Now().UTC().Add(-time.Hour)
+ q := &quantityStripe{sqid: fx.sqid, quantity: 10, status: "active", start: now}
+ ts := httptest.NewServer(http.HandlerFunc(q.serve))
+ t.Cleanup(ts.Close)
+ client, err := billingstripe.NewClient("rk_live_mock", billingstripe.WithBaseURL(ts.URL))
+ if err != nil {
+ t.Fatal(err)
+ }
+ fx.srv.billing = &billingRuntime{locks: newStoreWalkLocks(), client: client, cfg: BillingConfig{
+ MerchantAccount: "acct_carloku", TaxCode: "txcd_test", PortalConfiguration: "bpc_safe", LiveMode: true, Proration: proration,
+ CountryBands: map[string]string{"IE": "A"}, WebhookSecrets: []string{"whsec_test"},
+ Prices: map[string]string{billingPriceKey(plans.CurrentTermsVersion, plans.PlanCommunity, "A"): "price_community_A"},
+ }}
+ members, err := fx.st.Members(OpsAccountID)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for i := len(members); i < 10; i++ {
+ if err := fx.st.AddMember(OpsAccountID, fmt.Sprintf("m%d@example.test", i), RoleMember, now); err != nil {
+ t.Fatal(err)
+ }
+ }
+ state, err := fx.st.billingAccount(OpsAccountID)
+ if err != nil {
+ t.Fatal(err)
+ }
+ next := billingState{AccountID: OpsAccountID, Customer: "cus_team", Subscription: "sub_team", Status: "active", Plan: plans.PlanCommunity, Terms: plans.CurrentTermsVersion, Band: "A", Quantity: 10, PeriodStart: now.Unix(), PeriodEnd: now.AddDate(1, 0, 0).Unix(), LiveMode: true}
+ if err := fx.st.applyBillingState(next, state.Revision, "evt_quantity_seed", now); err != nil {
+ t.Fatal(err)
+ }
+ publishQuantityEntitlement(t, fx, now)
+ return fx, q, now
+}
+
+func publishQuantityEntitlement(t *testing.T, fx *consoleFixture, now time.Time) {
+ t.Helper()
+ row, err := scanEntitlementOutbox(fx.st.db.QueryRow(`SELECT `+entitlementOutboxColumns+` FROM entitlement_outbox WHERE account_id=?`, OpsAccountID))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := fx.st.markEntitlementPublished(OpsAccountID, row.Revision, row.Digest, now); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func quantityHeld(t *testing.T, fx *consoleFixture) bool {
+ t.Helper()
+ tx, err := fx.st.db.Begin()
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer tx.Rollback() //nolint:errcheck
+ held, err := quantityObligationTx(tx, OpsAccountID)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return held
+}
+
+func pendingAdditions(t *testing.T, fx *consoleFixture) []billingQuantityEvent {
+ t.Helper()
+ events, err := fx.st.billingQuantityEvents(OpsAccountID)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var out []billingQuantityEvent
+ for _, e := range events {
+ if e.Change == quantityEventAddition && e.ProcessedAt == 0 {
+ out = append(out, e)
+ }
+ }
+ return out
+}
+
+// With a policy configured, an added member raises the subscription to the
+// new quantity once, the billed quantity follows through reconcile, the
+// addition is marked, and the monthly freeze is no longer held.
+func TestQuantityWorkerRaisesTheSubscriptionAndClearsTheHold(t *testing.T) {
+ fx, q, now := newQuantityFixture(t, billingProrationImmediate)
+ ctx := context.Background()
+ if err := fx.st.AddMember(OpsAccountID, "eleventh@example.test", RoleMember, now.Add(time.Minute)); err != nil {
+ t.Fatal(err)
+ }
+ if !quantityHeld(t, fx) || len(pendingAdditions(t, fx)) != 1 {
+ t.Fatal("the addition should be pending and holding before the worker runs")
+ }
+ if err := fx.srv.billingQuantityPass(ctx, now.Add(2*time.Minute)); err != nil {
+ t.Fatal(err)
+ }
+ if len(q.posts) != 1 || q.posts[0]["quantity"] != "11" || q.posts[0]["proration_behavior"] != "always_invoice" || q.posts[0]["key"] == "" {
+ t.Fatalf("provider update: %+v", q.posts)
+ }
+ state, err := fx.st.billingAccount(OpsAccountID)
+ if err != nil || state.Quantity != 11 {
+ t.Fatalf("billed quantity %d %v, want 11", state.Quantity, err)
+ }
+ events, _ := fx.st.billingQuantityEvents(OpsAccountID)
+ last := events[len(events)-1]
+ if last.ProcessedAt == 0 || last.ProcessingOutcome != quantityOutcomeCharged || last.ProcessingReference != "si_team:quantity=11:always_invoice" {
+ t.Fatalf("event not marked as charged: %+v", last)
+ }
+ if quantityHeld(t, fx) {
+ t.Fatal("the freeze is still held after the addition was processed")
+ }
+ if sum, _ := fx.st.billingQuantityChange(OpsAccountID); sum.State != quantityChangeInSync || sum.BilledQuantity != 11 {
+ t.Fatalf("summary %+v", sum)
+ }
+ // A second pass has nothing to do and sends nothing.
+ if err := fx.srv.billingQuantityPass(ctx, now.Add(3*time.Minute)); err != nil {
+ t.Fatal(err)
+ }
+ if len(q.posts) != 1 {
+ t.Fatalf("a second pass sent again: %+v", q.posts)
+ }
+}
+
+// No policy, no provider call, no mark: the decision is not made for Paul.
+func TestQuantityWorkerDoesNothingWithoutAPolicy(t *testing.T) {
+ fx, q, now := newQuantityFixture(t, "")
+ if err := fx.st.AddMember(OpsAccountID, "eleventh@example.test", RoleMember, now.Add(time.Minute)); err != nil {
+ t.Fatal(err)
+ }
+ if err := fx.srv.billingQuantityPass(context.Background(), now.Add(2*time.Minute)); err != nil {
+ t.Fatal(err)
+ }
+ if q.gets != 0 || len(q.posts) != 0 || len(pendingAdditions(t, fx)) != 1 || !quantityHeld(t, fx) {
+ t.Fatalf("undecided policy acted: gets=%d posts=%v", q.gets, q.posts)
+ }
+}
+
+func TestQuantityWorkerNextInvoicePolicy(t *testing.T) {
+ fx, q, now := newQuantityFixture(t, billingProrationNextInvoice)
+ if err := fx.st.AddMember(OpsAccountID, "eleventh@example.test", RoleMember, now.Add(time.Minute)); err != nil {
+ t.Fatal(err)
+ }
+ if err := fx.srv.billingQuantityPass(context.Background(), now.Add(2*time.Minute)); err != nil {
+ t.Fatal(err)
+ }
+ if len(q.posts) != 1 || q.posts[0]["proration_behavior"] != "create_prorations" {
+ t.Fatalf("provider update: %+v", q.posts)
+ }
+ events, _ := fx.st.billingQuantityEvents(OpsAccountID)
+ if last := events[len(events)-1]; last.ProcessingOutcome != quantityOutcomeIncluded {
+ t.Fatalf("event outcome %q, want included", last.ProcessingOutcome)
+ }
+}
+
+// A seat added and removed inside the period is still a seat: removals
+// apply at renewal, so the target is the peak, not today's count.
+func TestQuantityWorkerChargesTheTransientSeat(t *testing.T) {
+ fx, q, now := newQuantityFixture(t, billingProrationImmediate)
+ if err := fx.st.AddMember(OpsAccountID, "brief@example.test", RoleMember, now.Add(time.Minute)); err != nil {
+ t.Fatal(err)
+ }
+ // RemoveMember stamps the wall clock, so the pass runs at the wall
+ // clock too: a pass dated before a recorded event is a clock reversal,
+ // which holds by design.
+ if err := fx.st.RemoveMember(OpsAccountID, "brief@example.test"); err != nil {
+ t.Fatal(err)
+ }
+ if err := fx.srv.billingQuantityPass(context.Background(), time.Now().UTC()); err != nil {
+ t.Fatal(err)
+ }
+ if len(q.posts) != 1 || q.posts[0]["quantity"] != "11" {
+ t.Fatalf("transient seat: %+v", q.posts)
+ }
+ if quantityHeld(t, fx) {
+ t.Fatal("still held")
+ }
+ if sum, _ := fx.st.billingQuantityChange(OpsAccountID); sum.State != quantityChangeAtRenewal {
+ t.Fatalf("summary %+v: ten members on eleven seats is a reduction at renewal", sum)
+ }
+}
+
+// The update landed but its response was lost. The next pass reads the
+// provider, sees the target already there, sends nothing and finishes.
+func TestQuantityWorkerRecoversALostResponse(t *testing.T) {
+ fx, q, now := newQuantityFixture(t, billingProrationImmediate)
+ ctx := context.Background()
+ if err := fx.st.AddMember(OpsAccountID, "eleventh@example.test", RoleMember, now.Add(time.Minute)); err != nil {
+ t.Fatal(err)
+ }
+ q.failAfter = true
+ if err := fx.srv.billingQuantityPass(ctx, now.Add(2*time.Minute)); err == nil {
+ t.Fatal("a lost response must surface as an error")
+ }
+ if len(pendingAdditions(t, fx)) != 1 || !quantityHeld(t, fx) {
+ t.Fatal("an unconfirmed update was marked processed")
+ }
+ if err := fx.srv.billingQuantityPass(ctx, now.Add(3*time.Minute)); err != nil {
+ t.Fatal(err)
+ }
+ if len(q.posts) != 1 {
+ t.Fatalf("recovery sent a second update: %+v", q.posts)
+ }
+ if state, _ := fx.st.billingAccount(OpsAccountID); state.Quantity != 11 || quantityHeld(t, fx) {
+ t.Fatalf("not recovered: quantity %d held %v", state.Quantity, quantityHeld(t, fx))
+ }
+}
+
+// A provider quantity the console neither billed nor wants is a person's
+// problem: a review is filed and nothing is sent.
+func TestQuantityWorkerRefusesAnUnexpectedProviderQuantity(t *testing.T) {
+ fx, q, now := newQuantityFixture(t, billingProrationImmediate)
+ if err := fx.st.AddMember(OpsAccountID, "eleventh@example.test", RoleMember, now.Add(time.Minute)); err != nil {
+ t.Fatal(err)
+ }
+ q.mu.Lock()
+ q.quantity = 13
+ q.mu.Unlock()
+ if err := fx.srv.billingQuantityPass(context.Background(), now.Add(2*time.Minute)); err != nil {
+ t.Fatal(err)
+ }
+ if len(q.posts) != 0 || len(pendingAdditions(t, fx)) != 1 {
+ t.Fatalf("acted on a mismatch: %+v", q.posts)
+ }
+ if rev, err := fx.st.billingReview(OpsAccountID, "sub_team", billingQuantityMismatchReason); err != nil || rev.ResolvedAt != 0 || rev.Occurrences != 1 {
+ t.Fatalf("no open review: %+v %v", rev, err)
+ }
+}
+
+// A failing subscription is never raised.
+func TestQuantityWorkerLeavesAPastDueSubscriptionAlone(t *testing.T) {
+ fx, q, now := newQuantityFixture(t, billingProrationImmediate)
+ if err := fx.st.AddMember(OpsAccountID, "eleventh@example.test", RoleMember, now.Add(time.Minute)); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := fx.st.db.Exec(`UPDATE billing_accounts SET status='past_due' WHERE account_id=?`, OpsAccountID); err != nil {
+ t.Fatal(err)
+ }
+ if err := fx.srv.billingQuantityPass(context.Background(), now.Add(2*time.Minute)); err != nil {
+ t.Fatal(err)
+ }
+ if q.gets != 0 || len(q.posts) != 0 || len(pendingAdditions(t, fx)) != 1 {
+ t.Fatalf("raised a past-due subscription: gets=%d posts=%v", q.gets, q.posts)
+ }
+}
diff --git a/internal/console/billingservice.go b/internal/console/billingservice.go
index 86c6cf54b..bc8018b83 100644
--- a/internal/console/billingservice.go
+++ b/internal/console/billingservice.go
@@ -27,6 +27,12 @@ type BillingConfig struct {
CountryBands map[string]string
HouseCoupon string
PortalConfiguration string
+ // Proration is the collection policy for a member added to a team plan
+ // mid-period (billingquantity.go): "immediate" invoices the prorated
+ // amount when the member is added, "next_invoice" adds it to the next
+ // invoice. Empty means the policy is undecided: nothing is sent to the
+ // provider and the additions stay pending, holding the monthly freeze.
+ Proration string
}
type billingRuntime struct {
@@ -60,6 +66,9 @@ func newBillingRuntime(cfg *BillingConfig) (*billingRuntime, error) {
return nil, errBillingUnavailable
}
}
+ if cfg.Proration != "" && cfg.Proration != billingProrationImmediate && cfg.Proration != billingProrationNextInvoice {
+ return nil, errBillingUnavailable
+ }
for country, band := range cfg.CountryBands {
if len(country) != 2 || strings.ToUpper(country) != country {
return nil, errBillingUnavailable