From 1a35f4eb91bf5f8bf79724f10d723b0c7dd340df Mon Sep 17 00:00:00 2001 From: Haitao Pan Date: Thu, 9 Apr 2026 13:50:55 +0800 Subject: [PATCH] feat(accounting): expose pricing-backed billing profile state --- api/accounting.go | 30 +++++++---- api/accounting_test.go | 38 ++++++++++++-- internal/store/accounting_memory.go | 38 ++++++++++++++ internal/store/accounting_postgres.go | 64 +++++++++++++++++++++++ internal/store/store.go | 18 +++++++ sql/20260401_accounting_control_plane.sql | 14 +++++ 6 files changed, 190 insertions(+), 12 deletions(-) diff --git a/api/accounting.go b/api/accounting.go index 53f9e37..94490c7 100644 --- a/api/accounting.go +++ b/api/accounting.go @@ -69,6 +69,7 @@ func (h *handler) accountUsageSummary(c *gin.Context) { suspendState := "active" throttleState := "normal" arrears := false + var billingProfile *store.AccountBillingProfile if quota, err := h.store.GetAccountQuotaState(c.Request.Context(), user.ID); err == nil && quota != nil { currentBalance = quota.CurrentBalance remainingQuota = quota.RemainingIncludedQuota @@ -76,6 +77,9 @@ func (h *handler) accountUsageSummary(c *gin.Context) { throttleState = quota.ThrottleState arrears = quota.Arrears } + if profile, err := h.store.GetAccountBillingProfile(c.Request.Context(), user.ID); err == nil && profile != nil { + billingProfile = profile + } syncDelaySeconds := 0 if lastBucketAt != nil { @@ -98,6 +102,7 @@ func (h *handler) accountUsageSummary(c *gin.Context) { "arrears": arrears, "lastBucketAt": lastBucketAt, "syncDelaySeconds": syncDelaySeconds, + "billingProfile": billingProfile, }) } @@ -147,12 +152,17 @@ func (h *handler) accountBillingSummary(c *gin.Context) { if snapshot, err := h.store.GetAccountQuotaState(c.Request.Context(), user.ID); err == nil { quota = snapshot } + var billingProfile *store.AccountBillingProfile + if profile, err := h.store.GetAccountBillingProfile(c.Request.Context(), user.ID); err == nil { + billingProfile = profile + } c.JSON(http.StatusOK, gin.H{ - "accountUuid": user.ID, - "quotaState": quota, - "ledger": ledger, - "sourceOfTruth": accountingDataSource, + "accountUuid": user.ID, + "quotaState": quota, + "billingProfile": billingProfile, + "ledger": ledger, + "sourceOfTruth": accountingDataSource, }) } @@ -218,13 +228,15 @@ func (h *handler) adminTrafficAccount(c *gin.Context) { } policy, _ := h.store.GetLatestAccountPolicySnapshot(c.Request.Context(), accountUUID) quota, _ := h.store.GetAccountQuotaState(c.Request.Context(), accountUUID) + billingProfile, _ := h.store.GetAccountBillingProfile(c.Request.Context(), accountUUID) c.JSON(http.StatusOK, gin.H{ - "accountUuid": accountUUID, - "buckets": buckets, - "ledger": ledger, - "policy": policy, - "quotaState": quota, + "accountUuid": accountUUID, + "buckets": buckets, + "ledger": ledger, + "policy": policy, + "quotaState": quota, + "billingProfile": billingProfile, }) } diff --git a/api/accounting_test.go b/api/accounting_test.go index e8172cb..849b8ce 100644 --- a/api/accounting_test.go +++ b/api/accounting_test.go @@ -138,6 +138,20 @@ func TestAccountUsageAndPolicyEndpoints(t *testing.T) { t.Fatalf("upsert quota state: %v", err) } + if err := st.UpsertAccountBillingProfile(ctx, &store.AccountBillingProfile{ + AccountUUID: user.ID, + PackageName: "starter", + IncludedQuotaBytes: 4096, + BasePricePerByte: 0.125, + RegionMultiplier: 1.2, + LineMultiplier: 1.5, + PeakMultiplier: 1.0, + OffPeakMultiplier: 1.0, + PricingRuleVersion: "pricing-v1", + }); err != nil { + t.Fatalf("upsert billing profile: %v", err) + } + if err := st.InsertBillingLedgerEntry(ctx, &store.BillingLedgerEntry{ ID: "ledger-1", AccountUUID: user.ID, @@ -179,9 +193,16 @@ func TestAccountUsageAndPolicyEndpoints(t *testing.T) { } var usagePayload struct { - AccountUUID string `json:"accountUuid"` - TotalBytes int64 `json:"totalBytes"` - SourceOfTruth string `json:"sourceOfTruth"` + AccountUUID string `json:"accountUuid"` + TotalBytes int64 `json:"totalBytes"` + SourceOfTruth string `json:"sourceOfTruth"` + BillingProfile struct { + PackageName string `json:"packageName"` + BasePricePerByte float64 `json:"basePricePerByte"` + RegionMultiplier float64 `json:"regionMultiplier"` + LineMultiplier float64 `json:"lineMultiplier"` + PricingRuleVersion string `json:"pricingRuleVersion"` + } `json:"billingProfile"` } if err := json.Unmarshal(rec.Body.Bytes(), &usagePayload); err != nil { t.Fatalf("decode usage payload: %v", err) @@ -195,6 +216,9 @@ func TestAccountUsageAndPolicyEndpoints(t *testing.T) { if usagePayload.SourceOfTruth != "postgresql" { t.Fatalf("expected source of truth postgresql, got %q", usagePayload.SourceOfTruth) } + if usagePayload.BillingProfile.PackageName != "starter" { + t.Fatalf("expected billing profile package starter, got %q", usagePayload.BillingProfile.PackageName) + } bucketsReq := httptest.NewRequest(http.MethodGet, "/api/account/usage/buckets", nil) bucketsReq.Header.Set("Authorization", "Bearer "+sessionToken) @@ -248,6 +272,11 @@ func TestAccountUsageAndPolicyEndpoints(t *testing.T) { QuotaState struct { CurrentBalance float64 `json:"currentBalance"` } `json:"quotaState"` + BillingProfile struct { + PackageName string `json:"packageName"` + IncludedQuotaBytes int64 `json:"includedQuotaBytes"` + BasePricePerByte float64 `json:"basePricePerByte"` + } `json:"billingProfile"` Ledger []struct { ID string `json:"id"` EntryType string `json:"entryType"` @@ -264,6 +293,9 @@ func TestAccountUsageAndPolicyEndpoints(t *testing.T) { if billingPayload.SourceOfTruth != "postgresql" { t.Fatalf("expected billing source of truth postgresql, got %q", billingPayload.SourceOfTruth) } + if billingPayload.BillingProfile.IncludedQuotaBytes != 4096 { + t.Fatalf("expected billing profile included quota 4096, got %d", billingPayload.BillingProfile.IncludedQuotaBytes) + } if billingPayload.QuotaState.CurrentBalance != 87.5 { t.Fatalf("expected billing current balance 87.5, got %v", billingPayload.QuotaState.CurrentBalance) } diff --git a/internal/store/accounting_memory.go b/internal/store/accounting_memory.go index 8e9ddf6..a7b11d6 100644 --- a/internal/store/accounting_memory.go +++ b/internal/store/accounting_memory.go @@ -61,6 +61,14 @@ func cloneQuotaState(src *AccountQuotaState) *AccountQuotaState { return © } +func cloneBillingProfile(src *AccountBillingProfile) *AccountBillingProfile { + if src == nil { + return nil + } + copy := *src + return © +} + func clonePolicySnapshot(src *AccountPolicySnapshot) *AccountPolicySnapshot { if src == nil { return nil @@ -265,6 +273,36 @@ func (s *memoryStore) GetAccountQuotaState(ctx context.Context, accountUUID stri return cloneQuotaState(record), nil } +func (s *memoryStore) UpsertAccountBillingProfile(ctx context.Context, profile *AccountBillingProfile) error { + _ = ctx + s.mu.Lock() + defer s.mu.Unlock() + + copy := cloneBillingProfile(profile) + if copy == nil { + return errors.New("billing profile is required") + } + now := time.Now().UTC() + if copy.CreatedAt.IsZero() { + copy.CreatedAt = now + } + copy.UpdatedAt = now + s.accountBillingProfiles[strings.TrimSpace(copy.AccountUUID)] = copy + return nil +} + +func (s *memoryStore) GetAccountBillingProfile(ctx context.Context, accountUUID string) (*AccountBillingProfile, error) { + _ = ctx + s.mu.RLock() + defer s.mu.RUnlock() + + record, ok := s.accountBillingProfiles[strings.TrimSpace(accountUUID)] + if !ok { + return nil, ErrUserNotFound + } + return cloneBillingProfile(record), nil +} + func (s *memoryStore) UpsertAccountPolicySnapshot(ctx context.Context, snapshot *AccountPolicySnapshot) error { _ = ctx s.mu.Lock() diff --git a/internal/store/accounting_postgres.go b/internal/store/accounting_postgres.go index 5f9413b..8bc116f 100644 --- a/internal/store/accounting_postgres.go +++ b/internal/store/accounting_postgres.go @@ -352,6 +352,70 @@ func (s *postgresStore) GetAccountQuotaState(ctx context.Context, accountUUID st return &state, nil } +func (s *postgresStore) UpsertAccountBillingProfile(ctx context.Context, profile *AccountBillingProfile) error { + if profile == nil { + return errors.New("billing profile is required") + } + + const query = ` + INSERT INTO account_billing_profiles ( + account_uuid, package_name, included_quota_bytes, base_price_per_byte, region_multiplier, line_multiplier, peak_multiplier, offpeak_multiplier, pricing_rule_version + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) + ON CONFLICT (account_uuid) DO UPDATE SET + package_name = EXCLUDED.package_name, + included_quota_bytes = EXCLUDED.included_quota_bytes, + base_price_per_byte = EXCLUDED.base_price_per_byte, + region_multiplier = EXCLUDED.region_multiplier, + line_multiplier = EXCLUDED.line_multiplier, + peak_multiplier = EXCLUDED.peak_multiplier, + offpeak_multiplier = EXCLUDED.offpeak_multiplier, + pricing_rule_version = EXCLUDED.pricing_rule_version, + updated_at = now() + RETURNING created_at, updated_at` + + return s.db.QueryRowContext( + ctx, + query, + strings.TrimSpace(profile.AccountUUID), + strings.TrimSpace(profile.PackageName), + profile.IncludedQuotaBytes, + profile.BasePricePerByte, + profile.RegionMultiplier, + profile.LineMultiplier, + profile.PeakMultiplier, + profile.OffPeakMultiplier, + strings.TrimSpace(profile.PricingRuleVersion), + ).Scan(&profile.CreatedAt, &profile.UpdatedAt) +} + +func (s *postgresStore) GetAccountBillingProfile(ctx context.Context, accountUUID string) (*AccountBillingProfile, error) { + const query = ` + SELECT account_uuid, package_name, included_quota_bytes, base_price_per_byte, region_multiplier, line_multiplier, peak_multiplier, offpeak_multiplier, pricing_rule_version, created_at, updated_at + FROM account_billing_profiles + WHERE account_uuid = $1` + var profile AccountBillingProfile + err := s.db.QueryRowContext(ctx, query, strings.TrimSpace(accountUUID)).Scan( + &profile.AccountUUID, + &profile.PackageName, + &profile.IncludedQuotaBytes, + &profile.BasePricePerByte, + &profile.RegionMultiplier, + &profile.LineMultiplier, + &profile.PeakMultiplier, + &profile.OffPeakMultiplier, + &profile.PricingRuleVersion, + &profile.CreatedAt, + &profile.UpdatedAt, + ) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return nil, ErrUserNotFound + } + return nil, err + } + return &profile, nil +} + func (s *postgresStore) UpsertAccountPolicySnapshot(ctx context.Context, snapshot *AccountPolicySnapshot) error { if snapshot == nil { return errors.New("policy snapshot is required") diff --git a/internal/store/store.go b/internal/store/store.go index ce3312f..afea724 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -131,6 +131,20 @@ type AccountQuotaState struct { UpdatedAt time.Time } +type AccountBillingProfile struct { + AccountUUID string + PackageName string + IncludedQuotaBytes int64 + BasePricePerByte float64 + RegionMultiplier float64 + LineMultiplier float64 + PeakMultiplier float64 + OffPeakMultiplier float64 + PricingRuleVersion string + CreatedAt time.Time + UpdatedAt time.Time +} + type AccountPolicySnapshot struct { AccountUUID string PolicyVersion string @@ -215,6 +229,8 @@ type Store interface { ListBillingLedgerByAccount(ctx context.Context, accountUUID string, limit int) ([]BillingLedgerEntry, error) UpsertAccountQuotaState(ctx context.Context, state *AccountQuotaState) error GetAccountQuotaState(ctx context.Context, accountUUID string) (*AccountQuotaState, error) + UpsertAccountBillingProfile(ctx context.Context, profile *AccountBillingProfile) error + GetAccountBillingProfile(ctx context.Context, accountUUID string) (*AccountBillingProfile, error) UpsertAccountPolicySnapshot(ctx context.Context, snapshot *AccountPolicySnapshot) error GetLatestAccountPolicySnapshot(ctx context.Context, accountUUID string) (*AccountPolicySnapshot, error) UpsertNodeHealthSnapshot(ctx context.Context, snapshot *NodeHealthSnapshot) error @@ -264,6 +280,7 @@ type memoryStore struct { trafficMinuteBuckets map[string]*TrafficMinuteBucket billingLedgerEntries map[string]*BillingLedgerEntry accountQuotaStates map[string]*AccountQuotaState + accountBillingProfiles map[string]*AccountBillingProfile accountPolicySnapshots map[string]*AccountPolicySnapshot nodeHealthSnapshots map[string]*NodeHealthSnapshot schedulerDecisions map[string]*SchedulerDecision @@ -309,6 +326,7 @@ func newMemoryStore(allowSuperAdminCounting bool) Store { trafficMinuteBuckets: make(map[string]*TrafficMinuteBucket), billingLedgerEntries: make(map[string]*BillingLedgerEntry), accountQuotaStates: make(map[string]*AccountQuotaState), + accountBillingProfiles: make(map[string]*AccountBillingProfile), accountPolicySnapshots: make(map[string]*AccountPolicySnapshot), nodeHealthSnapshots: make(map[string]*NodeHealthSnapshot), schedulerDecisions: make(map[string]*SchedulerDecision), diff --git a/sql/20260401_accounting_control_plane.sql b/sql/20260401_accounting_control_plane.sql index 73c835e..01f1135 100644 --- a/sql/20260401_accounting_control_plane.sql +++ b/sql/20260401_accounting_control_plane.sql @@ -53,6 +53,20 @@ CREATE TABLE IF NOT EXISTS public.account_quota_states ( updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ); +CREATE TABLE IF NOT EXISTS public.account_billing_profiles ( + account_uuid UUID PRIMARY KEY REFERENCES public.users(uuid) ON DELETE CASCADE, + package_name TEXT NOT NULL DEFAULT 'default', + included_quota_bytes BIGINT NOT NULL DEFAULT 0, + base_price_per_byte DOUBLE PRECISION NOT NULL DEFAULT 0, + region_multiplier DOUBLE PRECISION NOT NULL DEFAULT 1.0, + line_multiplier DOUBLE PRECISION NOT NULL DEFAULT 1.0, + peak_multiplier DOUBLE PRECISION NOT NULL DEFAULT 1.0, + offpeak_multiplier DOUBLE PRECISION NOT NULL DEFAULT 1.0, + pricing_rule_version TEXT NOT NULL DEFAULT 'pricing-default-v1', + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + CREATE TABLE IF NOT EXISTS public.account_policy_snapshots ( account_uuid UUID PRIMARY KEY REFERENCES public.users(uuid) ON DELETE CASCADE, policy_version TEXT NOT NULL,