diff --git a/server/db/queries/prices.sql b/server/db/queries/prices.sql index 782b115..0b15a69 100644 --- a/server/db/queries/prices.sql +++ b/server/db/queries/prices.sql @@ -13,26 +13,25 @@ SELECT * FROM model_prices WHERE model = ?; SELECT * FROM model_prices ORDER BY model ASC; -- name: ListModelStats :many --- Per-model aggregate: all-time token/spend totals + 48h TPS average. --- The since parameter scopes only the TPS calculation; token counts are --- all-time so spend reflects the full history. +-- Per-model aggregate from billing rows + potluck_requests. +-- since scopes the TPS window (48h); cost/token counts are all-time. SELECT - sp.model, + pkbr.model, COUNT(*) AS request_count, - SUM(sp.input_tokens) AS total_input_tokens, - SUM(sp.output_tokens) AS total_output_tokens, + SUM(COALESCE(pr.prompt_tokens, 0)) AS total_input_tokens, + SUM(COALESCE(pr.completion_tokens, 0)) AS total_output_tokens, AVG( CASE - WHEN st.finished_at IS NOT NULL - AND st.finished_at > st.started_at - AND st.started_at >= ? - THEN CAST(sp.output_tokens AS REAL) - / ((st.finished_at - st.started_at)) + WHEN pr.finished_at IS NOT NULL + AND pr.finished_at > pr.started_at + AND pr.started_at >= ? + THEN CAST(COALESCE(pr.completion_tokens, 0) AS REAL) + / (pr.finished_at - pr.started_at) ELSE NULL END ) AS avg_tps -FROM spends sp -JOIN streams st ON st.id = sp.stream_id -WHERE st.status = 'done' -GROUP BY sp.model +FROM pool_key_billing_rows pkbr +LEFT JOIN potluck_requests pr ON pr.id = pkbr.matched_request_id +WHERE pkbr.is_duplicate = 0 +GROUP BY pkbr.model ORDER BY request_count DESC; diff --git a/server/db/queries/usage.sql b/server/db/queries/usage.sql index 79db158..2599ec5 100644 --- a/server/db/queries/usage.sql +++ b/server/db/queries/usage.sql @@ -1,27 +1,35 @@ -- name: SpendByDay :many --- Daily spend for a user over the last N days. --- Returns one row per day (epoch of start of day in UTC) + amount in micros. +-- Daily spend for a user over the last N days, from billing rows. +-- Only potluck-routed rows (matched_request_id IS NOT NULL) are included. SELECT - (created_at / 86400) * 86400 AS day, - SUM(amount_micros) AS amount_micros, - SUM(input_tokens) AS input_tokens, - SUM(output_tokens) AS output_tokens -FROM spends -WHERE user_id = ? - AND created_at >= ? + (pkbr.pioneer_created_at / 86400) * 86400 AS day, + SUM(pkbr.cost_micros) AS amount_micros, + SUM(COALESCE(pr.prompt_tokens, 0)) AS input_tokens, + SUM(COALESCE(pr.completion_tokens, 0)) AS output_tokens +FROM pool_key_billing_rows pkbr +LEFT JOIN potluck_requests pr ON pr.id = pkbr.matched_request_id +WHERE pkbr.attributed_user_id = ? + AND pkbr.pioneer_created_at >= ? + AND pkbr.is_duplicate = 0 + AND pkbr.matched_request_id IS NOT NULL GROUP BY day -ORDER BY day ASC; +ORDER BY day ASC +; -- name: SpendByDayAndModel :many -- Daily spend broken down by model for a user, for stacked chart. SELECT - (created_at / 86400) * 86400 AS day, - model, - SUM(amount_micros) AS amount_micros, - SUM(input_tokens) AS input_tokens, - SUM(output_tokens) AS output_tokens -FROM spends -WHERE user_id = ? - AND created_at >= ? -GROUP BY day, model -ORDER BY day ASC, amount_micros DESC; + (pkbr.pioneer_created_at / 86400) * 86400 AS day, + pkbr.model, + SUM(pkbr.cost_micros) AS amount_micros, + SUM(COALESCE(pr.prompt_tokens, 0)) AS input_tokens, + SUM(COALESCE(pr.completion_tokens, 0)) AS output_tokens +FROM pool_key_billing_rows pkbr +LEFT JOIN potluck_requests pr ON pr.id = pkbr.matched_request_id +WHERE pkbr.attributed_user_id = ? + AND pkbr.pioneer_created_at >= ? + AND pkbr.is_duplicate = 0 + AND pkbr.matched_request_id IS NOT NULL +GROUP BY day, pkbr.model +ORDER BY day ASC, amount_micros DESC +; diff --git a/server/internal/api/v1/chat.go b/server/internal/api/v1/chat.go index 0550624..acbb5fb 100644 --- a/server/internal/api/v1/chat.go +++ b/server/internal/api/v1/chat.go @@ -9,6 +9,7 @@ import ( "net/http" "time" + charmlog "charm.land/log/v2" "github.com/google/uuid" apimw "github.com/taciturnaxolotl/potluck/internal/api/middleware" @@ -226,6 +227,16 @@ func (s *Server) streamCompletion(w http.ResponseWriter, r *http.Request, body [ } case e := <-errs: if e != nil { + charmlog.Error("stream error from pioneer", + "user_id", func() string { + if u != nil { return u.ID } + return "" + }(), + "model", model, + "pool_key_id", sel.KeyID(), + "req_id", reqID, + "err", e, + ) writeError(w, http.StatusBadGateway, "provider_error", e.Error()) if u != nil { go finishRequest(s.Q, reqID, 0, 0, 0, "error") diff --git a/server/internal/api/web/usage.go b/server/internal/api/web/usage.go index 3877403..c38134b 100644 --- a/server/internal/api/web/usage.go +++ b/server/internal/api/web/usage.go @@ -1,6 +1,7 @@ package web import ( + "database/sql" "net/http" "time" @@ -11,20 +12,20 @@ func (s *Server) handleUsage(w http.ResponseWriter, r *http.Request) { u, _ := currentUser(r) now := time.Now() - since30 := now.AddDate(0, 0, -30).Unix() + since30 := now.AddDate(0, 0, -29).Unix() daily, err := s.Q.SpendByDay(r.Context(), store.SpendByDayParams{ - UserID: u.ID, - CreatedAt: since30, + AttributedUserID: sql.NullString{String: u.ID, Valid: true}, + PioneerCreatedAt: since30, }) if err != nil { writeErr(w, 500, "internal", err.Error()) return } - since7 := now.AddDate(0, 0, -7).Unix() + since7 := now.AddDate(0, 0, -6).Unix() byModel, err := s.Q.SpendByDayAndModel(r.Context(), store.SpendByDayAndModelParams{ - UserID: u.ID, - CreatedAt: since7, + AttributedUserID: sql.NullString{String: u.ID, Valid: true}, + PioneerCreatedAt: since7, }) if err != nil { writeErr(w, 500, "internal", err.Error()) diff --git a/server/internal/provider/pioneer.go b/server/internal/provider/pioneer.go index 8be5e30..01dd04d 100644 --- a/server/internal/provider/pioneer.go +++ b/server/internal/provider/pioneer.go @@ -140,7 +140,12 @@ type rawChunk struct { } `json:"delta"` FinishReason *string `json:"finish_reason,omitempty"` } `json:"choices"` - Usage *Usage `json:"usage,omitempty"` + Usage *Usage `json:"usage,omitempty"` + Error *struct { + Message string `json:"message"` + Type string `json:"type"` + Code string `json:"code"` + } `json:"error,omitempty"` } func decodeChunk(payload []byte) (Chunk, error) { @@ -148,6 +153,11 @@ func decodeChunk(payload []byte) (Chunk, error) { if err := json.Unmarshal(payload, &rc); err != nil { return Chunk{}, fmt.Errorf("provider: decode chunk: %w", err) } + // Pioneer sends mid-stream errors as {"error":{"message":...}} chunks. + if rc.Error != nil { + return Chunk{}, fmt.Errorf("provider: upstream error: %s (type=%s code=%s)", + rc.Error.Message, rc.Error.Type, rc.Error.Code) + } c := Chunk{Raw: append([]byte(nil), payload...), Usage: rc.Usage} if len(rc.Choices) > 0 { c.Delta = rc.Choices[0].Delta.Content diff --git a/server/internal/store/prices.sql.go b/server/internal/store/prices.sql.go index addcfe9..74e9a29 100644 --- a/server/internal/store/prices.sql.go +++ b/server/internal/store/prices.sql.go @@ -60,24 +60,24 @@ func (q *Queries) ListModelPrices(ctx context.Context) ([]ModelPrice, error) { const listModelStats = `-- name: ListModelStats :many SELECT - sp.model, + pkbr.model, COUNT(*) AS request_count, - SUM(sp.input_tokens) AS total_input_tokens, - SUM(sp.output_tokens) AS total_output_tokens, + SUM(COALESCE(pr.prompt_tokens, 0)) AS total_input_tokens, + SUM(COALESCE(pr.completion_tokens, 0)) AS total_output_tokens, AVG( CASE - WHEN st.finished_at IS NOT NULL - AND st.finished_at > st.started_at - AND st.started_at >= ? - THEN CAST(sp.output_tokens AS REAL) - / ((st.finished_at - st.started_at)) + WHEN pr.finished_at IS NOT NULL + AND pr.finished_at > pr.started_at + AND pr.started_at >= ? + THEN CAST(COALESCE(pr.completion_tokens, 0) AS REAL) + / (pr.finished_at - pr.started_at) ELSE NULL END ) AS avg_tps -FROM spends sp -JOIN streams st ON st.id = sp.stream_id -WHERE st.status = 'done' -GROUP BY sp.model +FROM pool_key_billing_rows pkbr +LEFT JOIN potluck_requests pr ON pr.id = pkbr.matched_request_id +WHERE pkbr.is_duplicate = 0 +GROUP BY pkbr.model ORDER BY request_count DESC ` @@ -89,9 +89,8 @@ type ListModelStatsRow struct { AvgTps sql.NullFloat64 `json:"avg_tps"` } -// Per-model aggregate: all-time token/spend totals + 48h TPS average. -// The since parameter scopes only the TPS calculation; token counts are -// all-time so spend reflects the full history. +// Per-model aggregate from billing rows + potluck_requests. +// since scopes the TPS window (48h); cost/token counts are all-time. func (q *Queries) ListModelStats(ctx context.Context, startedAt int64) ([]ListModelStatsRow, error) { rows, err := q.db.QueryContext(ctx, listModelStats, startedAt) if err != nil { diff --git a/server/internal/store/querier.go b/server/internal/store/querier.go index 00639d7..6f5f23e 100644 --- a/server/internal/store/querier.go +++ b/server/internal/store/querier.go @@ -74,9 +74,8 @@ type Querier interface { ListMessagesForConversation(ctx context.Context, conversationID string) ([]Message, error) ListModelCatalog(ctx context.Context) ([]ModelsCatalog, error) ListModelPrices(ctx context.Context) ([]ModelPrice, error) - // Per-model aggregate: all-time token/spend totals + 48h TPS average. - // The since parameter scopes only the TPS calculation; token counts are - // all-time so spend reflects the full history. + // Per-model aggregate from billing rows + potluck_requests. + // since scopes the TPS window (48h); cost/token counts are all-time. ListModelStats(ctx context.Context, startedAt int64) ([]ListModelStatsRow, error) // Per-user pool key stats: shared contribution, today's spend, all-time spend. // Only users who have contributed at least one pool key appear. @@ -126,8 +125,8 @@ type Querier interface { RevokeAPIKey(ctx context.Context, arg RevokeAPIKeyParams) error SetPoolKeyActive(ctx context.Context, arg SetPoolKeyActiveParams) error SetStreamStatus(ctx context.Context, arg SetStreamStatusParams) error - // Daily spend for a user over the last N days. - // Returns one row per day (epoch of start of day in UTC) + amount in micros. + // Daily spend for a user over the last N days, from billing rows. + // Only potluck-routed rows (matched_request_id IS NOT NULL) are included. SpendByDay(ctx context.Context, arg SpendByDayParams) ([]SpendByDayRow, error) // Daily spend broken down by model for a user, for stacked chart. SpendByDayAndModel(ctx context.Context, arg SpendByDayAndModelParams) ([]SpendByDayAndModelRow, error) diff --git a/server/internal/store/usage.sql.go b/server/internal/store/usage.sql.go index af533f8..706235f 100644 --- a/server/internal/store/usage.sql.go +++ b/server/internal/store/usage.sql.go @@ -12,20 +12,23 @@ import ( const spendByDay = `-- name: SpendByDay :many SELECT - (created_at / 86400) * 86400 AS day, - SUM(amount_micros) AS amount_micros, - SUM(input_tokens) AS input_tokens, - SUM(output_tokens) AS output_tokens -FROM spends -WHERE user_id = ? - AND created_at >= ? + (pkbr.pioneer_created_at / 86400) * 86400 AS day, + SUM(pkbr.cost_micros) AS amount_micros, + SUM(COALESCE(pr.prompt_tokens, 0)) AS input_tokens, + SUM(COALESCE(pr.completion_tokens, 0)) AS output_tokens +FROM pool_key_billing_rows pkbr +LEFT JOIN potluck_requests pr ON pr.id = pkbr.matched_request_id +WHERE pkbr.attributed_user_id = ? + AND pkbr.pioneer_created_at >= ? + AND pkbr.is_duplicate = 0 + AND pkbr.matched_request_id IS NOT NULL GROUP BY day ORDER BY day ASC ` type SpendByDayParams struct { - UserID string `json:"user_id"` - CreatedAt int64 `json:"created_at"` + AttributedUserID sql.NullString `json:"attributed_user_id"` + PioneerCreatedAt int64 `json:"pioneer_created_at"` } type SpendByDayRow struct { @@ -35,10 +38,10 @@ type SpendByDayRow struct { OutputTokens sql.NullFloat64 `json:"output_tokens"` } -// Daily spend for a user over the last N days. -// Returns one row per day (epoch of start of day in UTC) + amount in micros. +// Daily spend for a user over the last N days, from billing rows. +// Only potluck-routed rows (matched_request_id IS NOT NULL) are included. func (q *Queries) SpendByDay(ctx context.Context, arg SpendByDayParams) ([]SpendByDayRow, error) { - rows, err := q.db.QueryContext(ctx, spendByDay, arg.UserID, arg.CreatedAt) + rows, err := q.db.QueryContext(ctx, spendByDay, arg.AttributedUserID, arg.PioneerCreatedAt) if err != nil { return nil, err } @@ -66,22 +69,27 @@ func (q *Queries) SpendByDay(ctx context.Context, arg SpendByDayParams) ([]Spend } const spendByDayAndModel = `-- name: SpendByDayAndModel :many +; + SELECT - (created_at / 86400) * 86400 AS day, - model, - SUM(amount_micros) AS amount_micros, - SUM(input_tokens) AS input_tokens, - SUM(output_tokens) AS output_tokens -FROM spends -WHERE user_id = ? - AND created_at >= ? -GROUP BY day, model + (pkbr.pioneer_created_at / 86400) * 86400 AS day, + pkbr.model, + SUM(pkbr.cost_micros) AS amount_micros, + SUM(COALESCE(pr.prompt_tokens, 0)) AS input_tokens, + SUM(COALESCE(pr.completion_tokens, 0)) AS output_tokens +FROM pool_key_billing_rows pkbr +LEFT JOIN potluck_requests pr ON pr.id = pkbr.matched_request_id +WHERE pkbr.attributed_user_id = ? + AND pkbr.pioneer_created_at >= ? + AND pkbr.is_duplicate = 0 + AND pkbr.matched_request_id IS NOT NULL +GROUP BY day, pkbr.model ORDER BY day ASC, amount_micros DESC ` type SpendByDayAndModelParams struct { - UserID string `json:"user_id"` - CreatedAt int64 `json:"created_at"` + AttributedUserID sql.NullString `json:"attributed_user_id"` + PioneerCreatedAt int64 `json:"pioneer_created_at"` } type SpendByDayAndModelRow struct { @@ -94,7 +102,7 @@ type SpendByDayAndModelRow struct { // Daily spend broken down by model for a user, for stacked chart. func (q *Queries) SpendByDayAndModel(ctx context.Context, arg SpendByDayAndModelParams) ([]SpendByDayAndModelRow, error) { - rows, err := q.db.QueryContext(ctx, spendByDayAndModel, arg.UserID, arg.CreatedAt) + rows, err := q.db.QueryContext(ctx, spendByDayAndModel, arg.AttributedUserID, arg.PioneerCreatedAt) if err != nil { return nil, err } diff --git a/web/src/routes/usage/+page.svelte b/web/src/routes/usage/+page.svelte index 1ef0209..c848a62 100644 --- a/web/src/routes/usage/+page.svelte +++ b/web/src/routes/usage/+page.svelte @@ -39,7 +39,7 @@ const now = Math.floor(Date.now() / 1000); const out: { day: number; usd: number }[] = []; for (let i = 0; i < 30; i++) { - const d = (Math.floor((now - 30 * 86400) / 86400) + i) * 86400; + const d = (Math.floor((now - 29 * 86400) / 86400) + i) * 86400; out.push({ day: d, usd: (map.get(d)?.amount_micros ?? 0) / 1_000_000 }); } return out; @@ -61,7 +61,7 @@ let stackData = $derived.by(() => { const now = Math.floor(Date.now() / 1000); const days = Array.from({ length: 7 }, (_, i) => - (Math.floor((now - 7 * 86400) / 86400) + i) * 86400 + (Math.floor((now - 6 * 86400) / 86400) + i) * 86400 ); const models = [...new Set(byModel.map(r => r.model))]; const lookup = new Map>(); @@ -126,9 +126,20 @@ let totalIn = $derived(daily.reduce((s, d) => s + d.input_tokens, 0)); let totalOut = $derived(daily.reduce((s, d) => s + d.output_tokens, 0)); - function fmtUSD(v: number) { if (v === 0) return '$0'; if (v < 0.001) return '<$0.001'; return '$' + v.toFixed(v < 0.1 ? 4 : 2); } + function fmtUSD(v: number) { + if (v === 0) return '$0'; + if (v < 0.001) return '<$0.001'; + if (v < 0.01) return '$' + v.toFixed(4); + if (v < 1) return '$' + v.toFixed(3).replace(/0+$/, '').replace(/\.$/, ''); + if (v < 100) return '$' + v.toFixed(2); + return '$' + Math.round(v).toLocaleString('en-US'); + } function fmtMicros(m: number) { return fmtUSD(m / 1_000_000); } - function fmtDay(unix: number) { return new Intl.DateTimeFormat('en-GB', { day: 'numeric', month: 'short' }).format(new Date(unix * 1000)); } + function fmtDay(unix: number) { + return new Intl.DateTimeFormat('en-GB', { + day: 'numeric', month: 'short', timeZone: 'UTC' + }).format(new Date(unix * 1000)); + } function fmtTokens(n: number) { if (n >= 1_000_000) return (n / 1_000_000).toFixed(1) + 'M'; if (n >= 1_000) return (n / 1_000).toFixed(0) + 'k';