usage

func (*CostDashboardAssembler) GetByToken ¶

func (a *CostDashboardAssembler) GetByToken(ctx context.Context, token string) (any, error)

GetByToken resolves the token, composes the usage view, sanitizes, and returns the public projection. Returns errs.ErrNotFound when the token doesn't match a customer — the handler surfaces this as 401 (anti-enumeration).

Empty-state contract: when the customer has no active subscription (no_subscription branch), the response carries empty arrays + a `billing_period.source = "no_subscription"` instead of a 5xx so the embed widget can render a clean empty state.

func (*CustomerUsageHandler) CustomerUsageRoutes ¶

func (h *CustomerUsageHandler) CustomerUsageRoutes(requireRead func(http.Handler) http.Handler) chi.Router

CustomerUsageRoutes returns the sub-router. Mount at /v1/customers/{id}/usage with the requireRead guard (auth.PermUsageRead); the customer ID is read from chi.URLParam(r, "id") after the sibling-mount, mirroring the /customers/{id}/coupon precedent.

func (*CustomerUsageService) Get ¶

func (s *CustomerUsageService) Get(ctx context.Context, tenantID, customerID string, period CustomerUsagePeriod) (CustomerUsageResult, error)

Get composes the customer-usage view. Order:

1. Customer existence (RLS makes cross-tenant IDs return ErrNotFound).
2. Active+trialing subscriptions for the customer.
3. Period resolution (current cycle from primary sub, or explicit ?from=&to= validated for partial bounds, ordering, 1-year cap).
4. Walk the meter union across subscribed plans, calling usage.AggregateByPricingRules per meter then ComputeAmountCents per rule. Same code path the cycle scan uses → dashboard math == invoice math.
5. Per-currency totals roll-up, warnings collection, subscription summary.

func (*Handler) Backfill ¶

func (h *Handler) Backfill(w http.ResponseWriter, r *http.Request)

Backfill is exported so the router can mount it behind PermUsageWrite — the /usage-events subtree uses PermUsageRead for historical reasons, but backfill is a sensitive ledger operation and should not be reachable by read-only keys.

func (*Handler) Routes ¶

func (h *Handler) Routes() chi.Router

func (*Handler) SummaryRoutes ¶

func (h *Handler) SummaryRoutes() chi.Router

SummaryHandler handles the usage summary HTTP endpoint.

func (*MarginAssembler) Get ¶

func (a *MarginAssembler) Get(ctx context.Context, tenantID, customerID string, from, to time.Time) (MarginReport, error)

func (*PostgresStore) Aggregate ¶

func (s *PostgresStore) Aggregate(ctx context.Context, filter ListFilter) (Aggregate, error)

Aggregate returns the totals + per-meter breakdown that powers the /usage page's stat cards + "Usage by Meter" card. It honours the same filter as List (customer / meter / from / to) but ignores limit/offset — the whole point is to surface filtered totals, not page slices.

SQL strategy: one row-of-totals query for COUNT/SUM/distinct-counts, plus a GROUP BY meter_id query for the per-meter breakdown. Both run inside the same tenant tx so RLS scopes them identically. SUM is decimal-precision (NUMERIC(38,12)) so 0.5 + 0.5 + 0.0001 round-trips to "1.0001" without floating-point drift.

func (*PostgresStore) AggregateByPricingRules ¶

func (s *PostgresStore) AggregateByPricingRules(
	ctx context.Context,
	tenantID, customerID, meterID string,
	defaultMode domain.AggregationMode,
	from, to time.Time,
) ([]domain.RuleAggregation, error)

AggregateByPricingRules implements the priority+claim resolution path described in docs/design-multi-dim-meters.md. Strategy:

1. Rank meter_pricing_rules by (priority DESC, created_at ASC, id ASC) so the ROW_NUMBER is fully deterministic. The id tiebreaker covers the corner case where two rules share the same priority AND were inserted within the same created_at tick (bulk import, same-txn bootstrap, clock-resolution collisions) — without it, Postgres is free to order them differently across servers, which would make billing non-reproducible across replicas.
2. LEFT JOIN LATERAL each in-period event against the ranked rules, keeping only the top-priority rule whose dimension_match is a subset of the event's properties. NULL rule means unclaimed.
3. Aggregate per rule. The CASE inside the SELECT is safe because every event in a given (rule_id) group shares the rule's mode — we GROUP BY (rule_id, mode, rrv) so the CASE evaluates a constant within each group.

last_ever needs a separate query because it ignores the period bounds. We run it only if any of the meter's rules is last_ever, then merge.

func (*PostgresStore) AggregateDailyBuckets ¶

func (s *PostgresStore) AggregateDailyBuckets(ctx context.Context, tenantID, customerID string, meterIDs []string, from, to time.Time) ([]DailyBucketRow, error)

AggregateDailyBuckets — see Store interface for contract. UTC-day granularity matches every reference platform (Datadog, OpenAI, AWS Cost Explorer); finer grain (hour) lives in a future bucket-grain param when an operator needs it. NULL meter_ids → empty result with no DB roundtrip. The result is unsorted; the service fills gaps and sorts by (bucket_start, meter_id) before serving.

func (*PostgresStore) AggregateForBillingPeriod ¶

func (s *PostgresStore) AggregateForBillingPeriod(ctx context.Context, tenantID, customerID string, meterIDs []string, from, to time.Time) (map[string]decimal.Decimal, error)

func (*PostgresStore) AggregateForBillingPeriodByAgg ¶

func (s *PostgresStore) AggregateForBillingPeriodByAgg(ctx context.Context, tenantID, customerID string, meters map[string]string, from, to time.Time) (map[string]decimal.Decimal, error)

func (*PostgresStore) CustomerProviderCost ¶

func (s *PostgresStore) CustomerProviderCost(ctx context.Context, tenantID, customerID string, from, to time.Time) (byModel []CustomerCostByModel, unresolvedEvents int64, err error)

func (*PostgresStore) DeleteProviderCostRate ¶

func (s *PostgresStore) DeleteProviderCostRate(ctx context.Context, tenantID, id string) error

DeleteProviderCostRate removes a rate row. Stamped events keep their snapshot (documented — deletion never rewrites history).

func (*PostgresStore) DeleteProviderCostRateAudited ¶ added in v0.2.0

func (s *PostgresStore) DeleteProviderCostRateAudited(
	ctx context.Context, tenantID, id string,
	emit func(tx *sql.Tx, deleted domain.ProviderCostRate) error,
) error

DeleteProviderCostRateAudited is DeleteProviderCostRate with an in-tx audit emission hook (ADR-090 shared fate). Two properties matter here:

• emit runs ONLY on a row that actually vanished. DELETE … RETURNING yields a row iff exactly one was removed; a miss is sql.ErrNoRows → errs.ErrNotFound with no emission, so deleting a nonexistent rate can never fabricate a "deleted" record.
• emit receives the DELETED row, read inside the tx. The row is gone afterwards, so the audit entry is the only surviving description of what the operator removed — an id alone would be unresolvable forever.

func (*PostgresStore) GetByIdempotencyKey ¶

func (s *PostgresStore) GetByIdempotencyKey(ctx context.Context, tenantID, key string) (domain.UsageEvent, error)

GetByIdempotencyKey fetches the event a replayed key originally wrote. Backs replay-as-success on the public ingest door: instead of a bare 409, the handler returns the original row (Stripe idempotency shape). Livemode scoping rides the RLS session like every other reader.

func (*PostgresStore) Ingest ¶

func (s *PostgresStore) Ingest(ctx context.Context, tenantID string, event domain.UsageEvent) (domain.UsageEvent, error)

func (*PostgresStore) IngestAudited ¶ added in v0.2.0

func (s *PostgresStore) IngestAudited(ctx context.Context, tenantID string, event domain.UsageEvent, emit func(tx *sql.Tx, out domain.UsageEvent) error) (domain.UsageEvent, error)

IngestAudited is Ingest with an in-tx audit emission hook (ADR-090).

Live ingest (POST /v1/usage-events, /batch, the LiteLLM spend callback) passes nil: it is machine metering at up to ~1000/s, usage_events IS the record, and a row per event would double the write volume of the hottest path for no forensic gain. BACKFILL is the exception and passes an emitter: an operator inserting BACKDATED usage is changing what a customer will be billed for a period that may already have closed — an operator action on the money path, not machine telemetry. Sharing the ingest tx means a backfilled event that cannot be recorded is not ingested at all.

func (*PostgresStore) IngestBatch ¶

func (s *PostgresStore) IngestBatch(ctx context.Context, tenantID string, events []domain.UsageEvent) (inserted, deduped int, err error)

IngestBatch writes every event in ONE transaction — all-or-nothing. Pre-fix, BatchIngest looped one tx per event: a mid-batch abort (client timeout, connection drop, crash) left a COMMITTED PREFIX, and the standard client response — retry the whole batch — re-ingested that prefix. Events without idempotency keys have no dedup line of defense, so the retry double-billed every prefix event. Atomic batches make the keyless retry safe: either the response arrived (don't retry) or nothing committed (retry is a clean first write).

Keyed duplicates (replay of a fully-committed batch, or the same key twice within one batch) are counted in deduped, not errors — matching the LiteLLM door's replay contract.

func (*PostgresStore) List ¶

func (s *PostgresStore) List(ctx context.Context, filter ListFilter) ([]domain.UsageEvent, int, error)

List paginates usage events ordered by (timestamp DESC, id DESC). Returns events + total (for the legacy offset path; 0 when cursor path used). Supports two mutually-exclusive paging shapes:

• Cursor (preferred, 2026-05-29): filter.AfterTimestamp + filter.AfterID set → seek-method query `WHERE (timestamp, id) < (after_ts, after_id)`. Skips the COUNT query, fetches limit+1 to detect hasMore via the next handler call. Stable across concurrent inserts at the table's head — usage_events is the highest-write table in Velox, so offset-based pagination page-skewed reliably whenever a cycle close fired between operator pages.
• Offset (legacy): COUNT + LIMIT/OFFSET. Kept for the dashboard paths that still use offset+total for "Page 1 of N" UX.

id as the tiebreaker keeps ordering deterministic when many events share a microsecond — common for batched ingestion.

func (*PostgresStore) ListProviderCostRates ¶

func (s *PostgresStore) ListProviderCostRates(ctx context.Context, tenantID string) ([]domain.ProviderCostRate, error)

ListProviderCostRates returns the tenant's current rates (mode-scoped via RLS), ordered for the dashboard table.

func (*PostgresStore) StreamForExport ¶ added in v0.2.0

func (s *PostgresStore) StreamForExport(ctx context.Context, tenantID string, from, to time.Time, fn func(domain.UsageEvent) error) error

StreamForExport walks EVERY usage event in the [from,to] window in ONE snapshot transaction. See customer.PostgresStore.StreamForExport for the general reason (issue #475); usage is the case that made it urgent. This table is the hottest write path in the product, and the export is given a five-minute timeout PRECISELY because it streams a lot of rows — which is exactly the window in which concurrent ingest is guaranteed. Under the old LIMIT/OFFSET-per-transaction paging, every event ingested mid-export shifted the newest-first window and duplicated the row on a page boundary. A duplicated usage row in a finance CSV reads as usage that did not happen.

The SELECT carries `origin`, which List's does not — so the `origin` column in usage-events.csv was always empty, and an operator could not tell an operator BACKFILL from a metered API event in the artifact they reconcile with. The export's columns are its own contract.

func (*PostgresStore) UpsertProviderCostRate ¶

func (s *PostgresStore) UpsertProviderCostRate(ctx context.Context, tenantID string, r domain.ProviderCostRate) (domain.ProviderCostRate, error)

UpsertProviderCostRate creates or edits-in-place the rate for a key (current-rate semantics, ADR-079 D1: the per-event stamp is the history; editing a rate only affects FUTURE events).

func (*PostgresStore) UpsertProviderCostRateAudited ¶ added in v0.2.0

func (s *PostgresStore) UpsertProviderCostRateAudited(
	ctx context.Context, tenantID string, r domain.ProviderCostRate,
	emit func(tx *sql.Tx, out domain.ProviderCostRate) error,
) (domain.ProviderCostRate, error)

UpsertProviderCostRateAudited is UpsertProviderCostRate with an in-tx audit emission hook: the rate write and its audit row commit or roll back together (ADR-090 shared fate). The store owns the transaction and exposes it to the closure; the caller (the handler — this surface has no service) owns row content. emit sees the PERSISTED rate, so the audit row carries the store-assigned id and the values as they actually landed. nil emit = unaudited upsert (unit-test / non-request callers).

The emission is unconditional on success by construction: an INSERT … ON CONFLICT DO UPDATE … RETURNING that scans a row always wrote one (a same-values re-PUT still bumps updated_at — a real mutation), so there is no zero-row arm to fabricate evidence for.

func (*ProviderCostHandler) Margin ¶

func (h *ProviderCostHandler) Margin(w http.ResponseWriter, r *http.Request)

Margin serves GET /v1/customers/{id}/margin?from&to (operator auth; mounted from the router next to the other customer subresources).

func (*ProviderCostHandler) Routes ¶

func (h *ProviderCostHandler) Routes() chi.Router

func (*ProviderCostHandler) SetAuditLogger ¶ added in v0.2.0

func (h *ProviderCostHandler) SetAuditLogger(a AuditEmitter)

SetAuditLogger wires in-tx audit emission for the rate mutations (ADR-090). There is no provider-cost service, so the handler builds the entry — it is the layer that knows the operator's intent — and hands it to the store's …Audited variants, which run it on the write's own transaction. A nil emitter skips emission (keeps handler unit tests fake-friendly); the composition root's audit.MustWired check is what makes a forgotten wiring line fail loudly at boot instead of silently un-auditing the routes.

func (*Service) Aggregate ¶

func (s *Service) Aggregate(ctx context.Context, filter ListFilter) (Aggregate, error)

Aggregate returns server-side totals + per-meter breakdown matching the filter. Used by GET /v1/usage-events/aggregate so the /usage dashboard's stat cards reflect filtered totals rather than a reduce over the current page of paginated events. Limit/Offset on the filter are intentionally ignored — the whole point is the unbounded total.

func (*Service) AggregateByPricingRules ¶

func (s *Service) AggregateByPricingRules(
	ctx context.Context,
	tenantID, customerID, meterID string,
	defaultMode domain.AggregationMode,
	from, to time.Time,
) ([]domain.RuleAggregation, error)

AggregateByPricingRules resolves a single (customer, meter, period) into per-rule aggregations using the priority+claim algorithm. The defaultMode applies to events that match no rule; it must be one of the four period-bounded modes (sum, count, max, last_during_period) — last_ever as a meter-default is rejected because it would silently break the "current state" semantics for unclaimed events.

See docs/design-multi-dim-meters.md for the resolution semantics; this method is the runtime entry point that billing-finalize will call.

func (*Service) AggregateDailyBuckets ¶

func (s *Service) AggregateDailyBuckets(ctx context.Context, tenantID, customerID string, meterIDs []string, from, to time.Time) ([]DailyBucketRow, error)

AggregateDailyBuckets delegates to the store. Exposed on the Service so CustomerUsageService can fetch chart data without holding the Store directly. See Store interface for contract.

func (*Service) AggregateForBillingPeriod ¶

func (s *Service) AggregateForBillingPeriod(ctx context.Context, tenantID, customerID string, meterIDs []string, from, to time.Time) (map[string]decimal.Decimal, error)

func (*Service) Backfill ¶

func (s *Service) Backfill(ctx context.Context, tenantID string, input IngestInput) (domain.UsageEvent, error)

Backfill ingests a historical usage event. Requires a non-nil timestamp strictly in the past; rejects missing / future / equal-to-now values so operators can't accidentally double-post a live event through the audit path. The row is tagged origin='backfill'.

Billing semantics: backfilled events participate in aggregation for any period whose [start, end) contains the event's timestamp. Finalized invoices are immutable (they reference billed_entries, not live aggregations), so backfill into closed periods is safe — it changes the audit ledger without rewriting history.

func (*Service) BatchIngest ¶

func (s *Service) BatchIngest(ctx context.Context, tenantID string, events []IngestInput) (int, int, []error)

BatchIngest validates every event, then writes them ALL in one store transaction — all-or-nothing. Pre-fix each event committed in its own tx: a mid-batch abort (client timeout, dropped connection) left a committed prefix, and the standard retry-the-batch response re-ingested that prefix — double-billed usage for every event without an idempotency key. Now either the whole batch lands or none of it does, so a keyless retry is always a clean first write.

Returns (inserted, deduped, errs). Validation failures are collected for EVERY failing index (a bare "quantity too large" across a 500-event batch was undebuggable) and abort before any write. Idempotency replays count as deduped — success, not failure — matching the LiteLLM door.

func (*Service) GetByIdempotencyKey ¶

func (s *Service) GetByIdempotencyKey(ctx context.Context, tenantID, key string) (domain.UsageEvent, error)

GetByIdempotencyKey fetches the event a replayed key originally wrote — backs the public door's replay-as-success response (200 + original row + Idempotent-Replayed header instead of a bare 409).

func (*Service) GetSummary ¶

func (s *Service) GetSummary(ctx context.Context, tenantID, customerID string, from, to time.Time) (UsageSummary, error)

GetSummary aggregates usage for a customer in the current billing period.

Aggregation is done server-side via store.Aggregate (COUNT(*) + GROUP BY meter_id SUM) over the full filtered set. The previous List(Limit:10000)+Go-reduce undercounted: List clamps the limit to 1000, so any customer with >1000 events in the window reported a truncated total — wrong on both the per-meter quantities and the event count.

func (*Service) Ingest ¶

func (s *Service) Ingest(ctx context.Context, tenantID string, input IngestInput) (domain.UsageEvent, error)

func (*Service) List ¶

func (s *Service) List(ctx context.Context, filter ListFilter) ([]domain.UsageEvent, int, error)

func (*Service) SetAuditLogger ¶ added in v0.2.0

func (s *Service) SetAuditLogger(a AuditEmitter)

SetAuditLogger wires in-tx audit emission for BACKFILL. Live ingest stays unaudited by design (machine metering; usage_events is the record) — but an operator inserting BACKDATED usage is changing what a customer gets billed for a period that may already have closed, so that action is recorded.

func (*Service) SetResolver ¶

func (s *Service) SetResolver(r clock.Resolver)

SetResolver wires the unified clock.Resolver (implemented by *billing.Engine). Ingest timestamps default to and gate against the CUSTOMER's effective now — a test clock's frozen_time when the customer is pinned — so usage ingestion works in simulated time: on a clock advanced into the (wall-clock) future, events without a timestamp land at frozen_time inside the simulated current period, and sim-timestamped events pass the future-skew gate that wall-clock comparison would wrongly reject. Optional: nil keeps wall-clock gating (narrow tests).