feat(analytics): native-Go /v1/analytics on datastore/ClickHouse, per-org
Adds clients/analytics (order 132, registered as analyticssvc so the real /v1/analytics/health owns the probe, not serve.go's generic liveness route) — the backend for the console Native Analytics module (unified-analytics.md §5). Two read lenses over the ONE hanzo warehouse, reusing the SAME clickhouse-go/v2 client the ai o11y ledger opens (ai/object DatastoreQuery/DatastoreEnabled/ EnsureCloudUsageTable/ResolveCloudUsageWindow) — no second CH client, DRY: - LLM lens (REAL): hanzo.cloud_usage — requests/tokens/spend/models/errorRate - web+commerce lens: hanzo.events — honest-empty until the collector emits Surface (read-only, org-scoped, /v1): GET /v1/analytics/overview per-org KPIs (llm real; web/commerce honest-empty) GET /v1/analytics/timeseries requests/tokens/spend over hour|day buckets GET /v1/analytics/top top models (real) + top products (honest-empty) GET /v1/analytics/health datastore connectivity + lens-table availability Tenant isolation is the security bar: tenant() requires a VALIDATED principal (c.User(), set by SanitizeIdentity only for a verified bearer) AND a valid org (c.Org(), the minted owner claim) — closing the Phase-1 no-bearer forged-X-Org-Id data path exactly as clients/s3 does. Every query binds the org POSITIONALLY (query.go llmWhere/eventsWhere), so a maxpower token can never read another org. ClickHouse creds are KMS-injected env (DATASTORE_*), never hardcoded. Tests: query-boundary isolation (org bound, never interpolated, incl SQLi slug), honest-empty, real-number KPIs, errorRate, gap-filled series, top-models pct; HTTP: no-principal->403, forged-org-no-bearer->403, datastore-down->honest 503, bad-range->400, health owned-by-analytics honest 503 when down.
This commit is contained in:
@@ -0,0 +1,374 @@
|
||||
// Copyright 2023-2026 Hanzo AI Inc. All Rights Reserved.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
// Package analytics mounts the Hanzo Cloud /v1/analytics/* surface: a native-Go,
|
||||
// per-org analytics read API over the `hanzo` ClickHouse warehouse (the
|
||||
// `datastore` cluster). It is the backend for the console Native Analytics module
|
||||
// (unified-analytics.md §5) — two read lenses over one warehouse:
|
||||
//
|
||||
// - LLM lens (REAL today): hanzo.cloud_usage, the live per-org usage ledger the
|
||||
// cloud o11y path already writes (requests, tokens, spend, models, errors).
|
||||
// - Web/commerce lens (honest-empty until the collector emits): hanzo.events.
|
||||
//
|
||||
// ONE ClickHouse client. This package does NOT open a second connection: it rides
|
||||
// the SAME clickhouse-go/v2 client the ai subsystem's o11y ledger opens in the
|
||||
// shared Bootstrap (ai/object.InitDatastore → object.DatastoreQuery). DRY: one
|
||||
// transport, one pool, one set of KMS-injected DATASTORE_* creds — never
|
||||
// hard-coded, never a second design.
|
||||
//
|
||||
// TENANT ISOLATION is the security bar and is enforced SERVER-SIDE on every
|
||||
// request. The org is c.Org() — the value SanitizeIdentity minted from the
|
||||
// VALIDATED bearer owner claim (HIP-0026), never a client header — AND every
|
||||
// request must carry a validated principal (c.User() set, which SanitizeIdentity
|
||||
// sets ONLY for a verified bearer). This closes the Phase-1 "no-bearer + forged
|
||||
// X-Org-Id direct-to-pod" cross-tenant read exactly as clients/s3 does. Every
|
||||
// ClickHouse query binds the org POSITIONALLY (query.go llmWhere/eventsWhere), so
|
||||
// a maxpower token can NEVER read another org's analytics.
|
||||
//
|
||||
// Surface (all org-scoped; /v1 only; read-only):
|
||||
//
|
||||
// GET /v1/analytics/overview per-org KPIs (llm real; web/commerce honest-empty)
|
||||
// GET /v1/analytics/timeseries requests/tokens/spend over time (hour|day buckets)
|
||||
// GET /v1/analytics/top top models (real) + top products (honest-empty)
|
||||
// GET /v1/analytics/health subsystem health (datastore connectivity + lens tables)
|
||||
//
|
||||
// Registered as "analyticssvc" (NOT "analytics") + order 132: the name diverges
|
||||
// from the /v1/analytics route prefix so serve.go's generic GET /v1/<name>/health
|
||||
// liveness route parks at /v1/analyticssvc/health and our REAL /v1/analytics/health
|
||||
// (below) owns the probe — the same health-shadow-avoidance the kmssvc/s3svc
|
||||
// subsystems use. Order 132 binds /v1/analytics/* before the ai subsystem's /v1/*
|
||||
// catch-all (150).
|
||||
package analytics
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
aiobject "github.com/hanzoai/ai/object"
|
||||
"github.com/hanzoai/cloud"
|
||||
"github.com/hanzoai/zip"
|
||||
luxlog "github.com/luxfi/log"
|
||||
)
|
||||
|
||||
const (
|
||||
// defaultTop / maxTop bound the /top result cardinality.
|
||||
defaultTop = 10
|
||||
maxTop = 100
|
||||
// probeTimeout bounds the health-endpoint table-existence probes so an
|
||||
// unauthenticated liveness hit can never hang on a slow warehouse.
|
||||
probeTimeout = 3 * time.Second
|
||||
)
|
||||
|
||||
type svc struct {
|
||||
log luxlog.Logger
|
||||
}
|
||||
|
||||
// Mount wires the analytics surface onto app per HIP-0106.
|
||||
func Mount(app *zip.App, deps cloud.Deps) error {
|
||||
if app == nil {
|
||||
return fmt.Errorf("analytics.Mount: nil zip.App")
|
||||
}
|
||||
log := deps.Logger
|
||||
if log == nil {
|
||||
return fmt.Errorf("analytics.Mount: nil deps.Logger")
|
||||
}
|
||||
log = log.New("subsystem", "analytics")
|
||||
s := &svc{log: log}
|
||||
|
||||
// Health owns /v1/analytics/health explicitly (not JWT-gated: liveness must be
|
||||
// probe-able). The data endpoints are all org-gated in-handler.
|
||||
app.Get("/v1/analytics/health", s.health)
|
||||
app.Get("/v1/analytics/overview", s.overview)
|
||||
app.Get("/v1/analytics/timeseries", s.timeseries)
|
||||
app.Get("/v1/analytics/top", s.top)
|
||||
|
||||
log.Info("analytics mounted", "warehouse", "hanzo", "brand", deps.Brand)
|
||||
return nil
|
||||
}
|
||||
|
||||
func init() {
|
||||
cloud.Register("analyticssvc", 132, func(app any, deps cloud.Deps) error {
|
||||
a, ok := app.(*zip.App)
|
||||
if !ok {
|
||||
return fmt.Errorf("analytics.Mount: app is %T, want *zip.App", app)
|
||||
}
|
||||
return Mount(a, deps)
|
||||
})
|
||||
}
|
||||
|
||||
// ── shared helpers ──────────────────────────────────────────────────────────
|
||||
|
||||
// tenant resolves the org — the tenant-isolation KEY — for a request, and refuses
|
||||
// the forgeable data path. It REQUIRES a validated principal: c.User() (X-User-Id)
|
||||
// is set by SanitizeIdentity ONLY when it verified a bearer/cookie; on the Phase-1
|
||||
// no-principal path it may RESTORE a client's raw X-Org-Id but leaves X-User-Id
|
||||
// empty. Gating on c.User() therefore refuses an in-cluster caller that forges
|
||||
// `X-Org-Id: victim` with NO bearer — the same defense clients/s3 uses — while
|
||||
// breaking no legitimate caller (all reach this via a user-bound bearer).
|
||||
//
|
||||
// The org is used EXACTLY as minted (no case-fold/normalize): the cloud_usage
|
||||
// ledger stored `organization` verbatim from the same owner claim, so an exact
|
||||
// match is required to see one's own rows (normalizing could collapse or miss).
|
||||
func tenant(c *zip.Ctx) (string, bool) {
|
||||
if strings.TrimSpace(c.User()) == "" {
|
||||
return "", false // no validated principal — refuse the forgeable data path
|
||||
}
|
||||
org := strings.TrimSpace(c.Org())
|
||||
if org == "" || len(org) > 128 {
|
||||
return "", false
|
||||
}
|
||||
return org, true
|
||||
}
|
||||
|
||||
// window resolves the [start,end) window + bucket interval from ?range/?start/?end,
|
||||
// reusing ai/object.ResolveCloudUsageWindow so analytics and the console2 Overview
|
||||
// share ONE window grammar (24h|7d|30d|custom). A bad range is a 400.
|
||||
func window(c *zip.Ctx) (time.Time, time.Time, string, string, error) {
|
||||
rangeLabel := strings.TrimSpace(c.Query("range"))
|
||||
start, end, interval, err := aiobject.ResolveCloudUsageWindow(rangeLabel, c.Query("start"), c.Query("end"), time.Now())
|
||||
if err != nil {
|
||||
return time.Time{}, time.Time{}, "", "", zip.ErrBadRequest(err.Error())
|
||||
}
|
||||
if rangeLabel == "" {
|
||||
rangeLabel = "24h"
|
||||
}
|
||||
return start, end, interval, rangeLabel, nil
|
||||
}
|
||||
|
||||
// requireDatastore returns the honest 503 when the ClickHouse ledger is not
|
||||
// connected, rather than fabricating zeros. Mirrors ai/object's read gate.
|
||||
func requireDatastore() error {
|
||||
if !aiobject.DatastoreEnabled() {
|
||||
return zip.Errorf(http.StatusServiceUnavailable, "analytics warehouse unavailable: datastore (ClickHouse) not connected")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func topLimit(c *zip.Ctx) int {
|
||||
n, err := strconv.Atoi(strings.TrimSpace(c.Query("limit")))
|
||||
if err != nil || n <= 0 {
|
||||
return defaultTop
|
||||
}
|
||||
if n > maxTop {
|
||||
return maxTop
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// ── /v1/analytics/overview ──────────────────────────────────────────────────
|
||||
|
||||
func (s *svc) overview(c *zip.Ctx) error {
|
||||
org, ok := tenant(c)
|
||||
if !ok {
|
||||
return zip.ErrForbidden("valid bearer required")
|
||||
}
|
||||
start, end, interval, rangeLabel, err := window(c)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := requireDatastore(); err != nil {
|
||||
return err
|
||||
}
|
||||
ctx := c.Context()
|
||||
// Ensure the ai-owned ledger table exists (idempotent, latched) so a fresh
|
||||
// warehouse yields honest zeros, not an error. We NEVER create hanzo.events —
|
||||
// that table is operator-owned (unified-analytics.md §3.1).
|
||||
if err := aiobject.EnsureCloudUsageTable(ctx); err != nil {
|
||||
return zip.Errorf(http.StatusServiceUnavailable, "analytics warehouse unavailable: %v", err)
|
||||
}
|
||||
|
||||
// LLM lens — REAL per-org KPIs.
|
||||
where, args := llmWhere(org, start, end)
|
||||
llmSQL := "SELECT count() AS requests, sum(total_tokens) AS tokens, " +
|
||||
"sum(prompt_tokens) AS prompt_tokens, sum(completion_tokens) AS completion_tokens, " +
|
||||
"sum(cost_cents) AS cost_cents, uniqExact(model) AS models, uniqExact(provider) AS providers, " +
|
||||
"countIf(status = 'error') AS errors FROM " + llmTable + " WHERE " + where
|
||||
llmRows, err := aiobject.DatastoreQuery(ctx, llmSQL, args...)
|
||||
if err != nil {
|
||||
return zip.Errorf(http.StatusBadGateway, "analytics llm query: %v", err)
|
||||
}
|
||||
llm := buildLLMOverview(firstRow(llmRows))
|
||||
|
||||
// Web/commerce lens — one events query; degrades to honest-empty if the events
|
||||
// table is absent (not yet provisioned) or errors.
|
||||
ewhere, eargs := eventsWhere(org, start, end)
|
||||
eventsSQL := "SELECT countIf(event = '$pageview') AS pageviews, uniqExact(distinct_id) AS visitors, " +
|
||||
"uniqExact(session_id) AS sessions, countIf(event = 'order_completed') AS orders, " +
|
||||
"toFloat64(sum(revenue)) AS revenue FROM " + eventsTable + " WHERE " + ewhere
|
||||
eventsRows, eerr := aiobject.DatastoreQuery(ctx, eventsSQL, eargs...)
|
||||
eventsOK := eerr == nil
|
||||
if eerr != nil {
|
||||
s.log.Debug("events lens unavailable (honest-empty)", "err", eerr)
|
||||
}
|
||||
erow := firstRow(eventsRows)
|
||||
|
||||
return c.JSON(http.StatusOK, Overview{
|
||||
Range: rangeLabel,
|
||||
Start: start.UTC().Format(time.RFC3339),
|
||||
End: end.UTC().Format(time.RFC3339),
|
||||
Interval: interval,
|
||||
Scope: Scope{Org: org},
|
||||
LLM: llm,
|
||||
Web: buildWebOverview(erow, eventsOK),
|
||||
Commerce: buildCommerceOverview(erow, eventsOK),
|
||||
})
|
||||
}
|
||||
|
||||
// ── /v1/analytics/timeseries ────────────────────────────────────────────────
|
||||
|
||||
func (s *svc) timeseries(c *zip.Ctx) error {
|
||||
org, ok := tenant(c)
|
||||
if !ok {
|
||||
return zip.ErrForbidden("valid bearer required")
|
||||
}
|
||||
start, end, interval, rangeLabel, err := window(c)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := requireDatastore(); err != nil {
|
||||
return err
|
||||
}
|
||||
ctx := c.Context()
|
||||
if err := aiobject.EnsureCloudUsageTable(ctx); err != nil {
|
||||
return zip.Errorf(http.StatusServiceUnavailable, "analytics warehouse unavailable: %v", err)
|
||||
}
|
||||
|
||||
// bucketFn is a CLOSED server-chosen enum (never user input), so interpolating
|
||||
// it is injection-safe; the org + time bounds stay bound parameters.
|
||||
bucketFn := "Hour"
|
||||
if interval == "day" {
|
||||
bucketFn = "Day"
|
||||
}
|
||||
where, args := llmWhere(org, start, end)
|
||||
seriesSQL := fmt.Sprintf("SELECT toStartOf%s(timestamp, 'UTC') AS bucket, count() AS requests, "+
|
||||
"sum(total_tokens) AS tokens, sum(cost_cents) AS cost_cents FROM %s WHERE %s GROUP BY bucket ORDER BY bucket",
|
||||
bucketFn, llmTable, where)
|
||||
rows, err := aiobject.DatastoreQuery(ctx, seriesSQL, args...)
|
||||
if err != nil {
|
||||
return zip.Errorf(http.StatusBadGateway, "analytics timeseries query: %v", err)
|
||||
}
|
||||
|
||||
return c.JSON(http.StatusOK, Timeseries{
|
||||
Range: rangeLabel,
|
||||
Start: start.UTC().Format(time.RFC3339),
|
||||
End: end.UTC().Format(time.RFC3339),
|
||||
Interval: interval,
|
||||
Scope: Scope{Org: org},
|
||||
Series: buildSeries(start, end, interval, rows),
|
||||
Source: llmTable,
|
||||
})
|
||||
}
|
||||
|
||||
// ── /v1/analytics/top ───────────────────────────────────────────────────────
|
||||
|
||||
func (s *svc) top(c *zip.Ctx) error {
|
||||
org, ok := tenant(c)
|
||||
if !ok {
|
||||
return zip.ErrForbidden("valid bearer required")
|
||||
}
|
||||
start, end, _, rangeLabel, err := window(c)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := requireDatastore(); err != nil {
|
||||
return err
|
||||
}
|
||||
ctx := c.Context()
|
||||
if err := aiobject.EnsureCloudUsageTable(ctx); err != nil {
|
||||
return zip.Errorf(http.StatusServiceUnavailable, "analytics warehouse unavailable: %v", err)
|
||||
}
|
||||
limit := topLimit(c)
|
||||
|
||||
// Top models — REAL. limit is a validated int (never user text) so %d is safe;
|
||||
// org + time stay bound parameters.
|
||||
where, args := llmWhere(org, start, end)
|
||||
modelSQL := fmt.Sprintf("SELECT model, any(provider) AS provider, count() AS requests, "+
|
||||
"sum(total_tokens) AS tokens, sum(cost_cents) AS cost_cents FROM %s WHERE %s "+
|
||||
"GROUP BY model ORDER BY cost_cents DESC, requests DESC LIMIT %d", llmTable, where, limit)
|
||||
modelRows, err := aiobject.DatastoreQuery(ctx, modelSQL, args...)
|
||||
if err != nil {
|
||||
return zip.Errorf(http.StatusBadGateway, "analytics top-models query: %v", err)
|
||||
}
|
||||
|
||||
// Top products — honest-empty until commerce emits order events.
|
||||
ewhere, eargs := eventsWhere(org, start, end)
|
||||
prodSQL := fmt.Sprintf("SELECT product_id AS productId, countIf(event = 'order_completed') AS orders, "+
|
||||
"toFloat64(sum(revenue)) AS revenue, sum(quantity) AS units FROM %s WHERE %s AND product_id != '' "+
|
||||
"GROUP BY product_id ORDER BY revenue DESC LIMIT %d", eventsTable, ewhere, limit)
|
||||
prodRows, perr := aiobject.DatastoreQuery(ctx, prodSQL, eargs...)
|
||||
|
||||
return c.JSON(http.StatusOK, Top{
|
||||
Range: rangeLabel,
|
||||
Start: start.UTC().Format(time.RFC3339),
|
||||
End: end.UTC().Format(time.RFC3339),
|
||||
Scope: Scope{Org: org},
|
||||
Models: buildTopModels(modelRows),
|
||||
Products: buildTopProducts(prodRows, perr == nil),
|
||||
})
|
||||
}
|
||||
|
||||
// ── /v1/analytics/health ────────────────────────────────────────────────────
|
||||
|
||||
// health is a REAL probe: it reports datastore connectivity (the load-bearing
|
||||
// signal) and, when connected, the availability of each lens table. Not
|
||||
// JWT-gated (liveness must be probe-able) and it NEVER reads tenant data — only
|
||||
// table existence. 503 when the warehouse is unreachable so a readiness probe
|
||||
// can gate; 200 otherwise even if the events lens is not yet provisioned (that is
|
||||
// honest-empty, not a failure).
|
||||
func (s *svc) health(c *zip.Ctx) error {
|
||||
connected := aiobject.DatastoreEnabled()
|
||||
res := map[string]any{
|
||||
"service": "analytics",
|
||||
"status": "ok",
|
||||
"datastore": connected,
|
||||
"warehouse": "hanzo",
|
||||
}
|
||||
if !connected {
|
||||
res["status"] = "degraded"
|
||||
res["reason"] = "datastore (ClickHouse) not connected"
|
||||
return c.JSON(http.StatusServiceUnavailable, res)
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(c.Context(), probeTimeout)
|
||||
defer cancel()
|
||||
res["lenses"] = map[string]any{
|
||||
"llm": map[string]any{"table": llmTable, "available": tableExists(ctx, llmTable)},
|
||||
"events": map[string]any{"table": eventsTable, "available": tableExists(ctx, eventsTable)},
|
||||
}
|
||||
return c.JSON(http.StatusOK, res)
|
||||
}
|
||||
|
||||
// tableExists probes ClickHouse for a table's presence. The name is a package
|
||||
// constant (never user input), so `EXISTS TABLE` is safe. Any error → false
|
||||
// (honest "not available") rather than surfacing.
|
||||
func tableExists(ctx context.Context, qualified string) bool {
|
||||
rows, err := aiobject.DatastoreQuery(ctx, "EXISTS TABLE "+qualified)
|
||||
if err != nil || len(rows) == 0 {
|
||||
return false
|
||||
}
|
||||
for _, v := range rows[0] {
|
||||
return aInt64(v) == 1
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func firstRow(rows []map[string]any) map[string]any {
|
||||
if len(rows) == 0 {
|
||||
return map[string]any{}
|
||||
}
|
||||
return rows[0]
|
||||
}
|
||||
@@ -0,0 +1,185 @@
|
||||
// Copyright 2023-2026 Hanzo AI Inc. All Rights Reserved.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package analytics
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// TestLLMWhereBindsOrgPositionally is THE tenant-isolation proof at the SQL
|
||||
// boundary: the org is ALWAYS the trailing bound parameter (never interpolated),
|
||||
// the predicate is "organization = ?", and the org value NEVER appears in the SQL
|
||||
// string. So a maxpower query and an acme query differ ONLY in a bound arg — one
|
||||
// tenant can never read another's rows, and a hostile org slug can't escape into
|
||||
// SQL.
|
||||
func TestLLMWhereBindsOrgPositionally(t *testing.T) {
|
||||
start := time.Date(2026, 6, 24, 0, 0, 0, 0, time.UTC)
|
||||
end := time.Date(2026, 7, 1, 0, 0, 0, 0, time.UTC)
|
||||
|
||||
for _, org := range []string{"maxpower", "acme", "o'; DROP TABLE hanzo.cloud_usage; --"} {
|
||||
sql, args := llmWhere(org, start, end)
|
||||
if !strings.Contains(sql, "organization = ?") {
|
||||
t.Fatalf("llmWhere sql must bind organization: %q", sql)
|
||||
}
|
||||
if strings.Contains(sql, org) {
|
||||
t.Fatalf("org %q must NOT be interpolated into sql: %q", org, sql)
|
||||
}
|
||||
if len(args) != 3 {
|
||||
t.Fatalf("want 3 bound args (start,end,org), got %d: %v", len(args), args)
|
||||
}
|
||||
if got, ok := args[2].(string); !ok || got != org {
|
||||
t.Fatalf("org must be the trailing bound arg verbatim, want %q got %v", org, args[2])
|
||||
}
|
||||
// Time bounds are also bound (as CH DateTime literals), never interpolated.
|
||||
if !strings.Contains(sql, "timestamp >= ? AND timestamp < ?") {
|
||||
t.Fatalf("time bounds must be parameterized: %q", sql)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestEventsWhereBindsOrgPositionally: the events lens keys on tenant_id, same
|
||||
// bound-parameter discipline.
|
||||
func TestEventsWhereBindsOrgPositionally(t *testing.T) {
|
||||
start := time.Date(2026, 6, 24, 0, 0, 0, 0, time.UTC)
|
||||
end := time.Date(2026, 7, 1, 0, 0, 0, 0, time.UTC)
|
||||
sql, args := eventsWhere("maxpower", start, end)
|
||||
if !strings.Contains(sql, "tenant_id = ?") {
|
||||
t.Fatalf("eventsWhere must bind tenant_id: %q", sql)
|
||||
}
|
||||
if strings.Contains(sql, "maxpower") {
|
||||
t.Fatalf("org must not be interpolated: %q", sql)
|
||||
}
|
||||
if got, ok := args[2].(string); !ok || got != "maxpower" {
|
||||
t.Fatalf("org must be trailing bound arg, got %v", args[2])
|
||||
}
|
||||
}
|
||||
|
||||
// TestBuildLLMOverviewRealNumbers: the flagship assembler over a realistic row
|
||||
// (maxpower's live shape ≈ 21 req / 3.2K tokens / $1.20 / 3 models). Proves the
|
||||
// KPIs and the errorRate math are exact.
|
||||
func TestBuildLLMOverviewRealNumbers(t *testing.T) {
|
||||
// Mimics the direct ClickHouse driver's native scan types (uint64 aggregates).
|
||||
row := map[string]any{
|
||||
"requests": uint64(21),
|
||||
"tokens": uint64(3200),
|
||||
"prompt_tokens": uint64(2100),
|
||||
"completion_tokens": uint64(1100),
|
||||
"cost_cents": uint64(120),
|
||||
"models": uint64(3),
|
||||
"providers": uint64(2),
|
||||
"errors": uint64(0),
|
||||
}
|
||||
o := buildLLMOverview(row)
|
||||
if !o.Available {
|
||||
t.Fatal("llm overview must be available when the datastore answered")
|
||||
}
|
||||
if o.Requests != 21 || o.Tokens != 3200 || o.SpendCents != 120 || o.Models != 3 || o.Providers != 2 {
|
||||
t.Fatalf("KPI mismatch: %+v", o)
|
||||
}
|
||||
if o.PromptTokens != 2100 || o.CompletionTokens != 1100 {
|
||||
t.Fatalf("token split mismatch: %+v", o)
|
||||
}
|
||||
if o.ErrorRate != 0 {
|
||||
t.Fatalf("errorRate want 0, got %v", o.ErrorRate)
|
||||
}
|
||||
if o.Source != "hanzo.cloud_usage" {
|
||||
t.Fatalf("source want hanzo.cloud_usage, got %q", o.Source)
|
||||
}
|
||||
}
|
||||
|
||||
// TestBuildLLMOverviewHonestEmpty: an empty aggregate (no usage in the window)
|
||||
// yields honest zeros — never fabricated, and NOT unavailable (the datastore did
|
||||
// answer; there is just nothing).
|
||||
func TestBuildLLMOverviewHonestEmpty(t *testing.T) {
|
||||
o := buildLLMOverview(map[string]any{})
|
||||
if !o.Available {
|
||||
t.Fatal("empty window must still be Available (honest-zero, not unavailable)")
|
||||
}
|
||||
if o.Requests != 0 || o.Tokens != 0 || o.SpendCents != 0 || o.Models != 0 || o.ErrorRate != 0 {
|
||||
t.Fatalf("empty overview must be all-zero, got %+v", o)
|
||||
}
|
||||
}
|
||||
|
||||
// TestErrorRate: errors/requests, rounded to 3 places.
|
||||
func TestErrorRate(t *testing.T) {
|
||||
o := buildLLMOverview(map[string]any{"requests": uint64(10), "errors": uint64(2)})
|
||||
if o.ErrorRate != 0.2 {
|
||||
t.Fatalf("errorRate want 0.2, got %v", o.ErrorRate)
|
||||
}
|
||||
}
|
||||
|
||||
// TestOrgAOverviewDiffersFromOrgB: combined with the where-isolation proof, each
|
||||
// org's query returns only its own rows; distinct rows assemble to distinct
|
||||
// overviews. (org A's overview != org B's.)
|
||||
func TestOrgAOverviewDiffersFromOrgB(t *testing.T) {
|
||||
a := buildLLMOverview(map[string]any{"requests": uint64(21), "tokens": uint64(3200), "cost_cents": uint64(120)})
|
||||
b := buildLLMOverview(map[string]any{"requests": uint64(4), "tokens": uint64(500), "cost_cents": uint64(9)})
|
||||
if a.Requests == b.Requests || a.Tokens == b.Tokens || a.SpendCents == b.SpendCents {
|
||||
t.Fatalf("distinct orgs must assemble distinct overviews: a=%+v b=%+v", a, b)
|
||||
}
|
||||
}
|
||||
|
||||
// TestBuildSeriesGapFill: sparse ClickHouse buckets become an evenly-spaced,
|
||||
// gap-filled series across the window (zeros where no data, real where present).
|
||||
func TestBuildSeriesGapFill(t *testing.T) {
|
||||
start := time.Date(2026, 6, 28, 0, 0, 0, 0, time.UTC)
|
||||
end := time.Date(2026, 7, 1, 0, 0, 0, 0, time.UTC) // 3 daily buckets: 28, 29, 30
|
||||
rows := []map[string]any{
|
||||
{"bucket": time.Date(2026, 6, 29, 0, 0, 0, 0, time.UTC), "requests": uint64(5), "tokens": uint64(100), "cost_cents": uint64(10)},
|
||||
}
|
||||
series := buildSeries(start, end, "day", rows)
|
||||
if len(series) != 3 {
|
||||
t.Fatalf("want 3 gap-filled daily points, got %d: %+v", len(series), series)
|
||||
}
|
||||
if series[0].Requests != 0 || series[0].T != "2026-06-28T00:00:00Z" {
|
||||
t.Fatalf("first bucket must be honest-zero 06-28, got %+v", series[0])
|
||||
}
|
||||
if series[1].Requests != 5 || series[1].Tokens != 100 || series[1].SpendCents != 10 {
|
||||
t.Fatalf("06-29 bucket must carry real data, got %+v", series[1])
|
||||
}
|
||||
if series[2].Requests != 0 {
|
||||
t.Fatalf("06-30 bucket must be honest-zero, got %+v", series[2])
|
||||
}
|
||||
}
|
||||
|
||||
// TestBuildTopModelsSortAndPct: models sort by spend desc and each pct is its
|
||||
// share of total spend.
|
||||
func TestBuildTopModelsSortAndPct(t *testing.T) {
|
||||
rows := []map[string]any{
|
||||
{"model": "gpt-4o-mini", "provider": "do-ai", "requests": uint64(3), "tokens": uint64(200), "cost_cents": uint64(20)},
|
||||
{"model": "claude-sonnet-4-5", "provider": "anthropic", "requests": uint64(9), "tokens": uint64(3000), "cost_cents": uint64(80)},
|
||||
}
|
||||
top := buildTopModels(rows)
|
||||
if !top.Available || len(top.Items) != 2 {
|
||||
t.Fatalf("want 2 models available, got %+v", top)
|
||||
}
|
||||
if top.Items[0].Model != "claude-sonnet-4-5" {
|
||||
t.Fatalf("highest-spend model must sort first, got %q", top.Items[0].Model)
|
||||
}
|
||||
// 80 of 100 total = 80%, 20 of 100 = 20%.
|
||||
if top.Items[0].Pct != 80 || top.Items[1].Pct != 20 {
|
||||
t.Fatalf("pct shares wrong: %v / %v", top.Items[0].Pct, top.Items[1].Pct)
|
||||
}
|
||||
}
|
||||
|
||||
// TestBuildTopProductsHonestEmpty: when the events table is absent (ok=false) the
|
||||
// products lens is honestly reported unavailable with an empty (non-nil) list.
|
||||
func TestBuildTopProductsHonestEmpty(t *testing.T) {
|
||||
tp := buildTopProducts(nil, false)
|
||||
if tp.Available {
|
||||
t.Fatal("products must be unavailable when events table is absent")
|
||||
}
|
||||
if tp.Items == nil || len(tp.Items) != 0 {
|
||||
t.Fatalf("items must be an empty (non-nil) slice, got %#v", tp.Items)
|
||||
}
|
||||
if tp.Reason == "" || tp.Source != "hanzo.events" {
|
||||
t.Fatalf("must carry honest reason + source, got %+v", tp)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,131 @@
|
||||
// Copyright 2023-2026 Hanzo AI Inc. All Rights Reserved.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package analytics
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"github.com/hanzoai/cloud"
|
||||
"github.com/hanzoai/zip"
|
||||
luxlog "github.com/luxfi/log"
|
||||
)
|
||||
|
||||
func mountApp(t *testing.T) *zip.App {
|
||||
t.Helper()
|
||||
app := zip.New(zip.Config{Logger: luxlog.New("test")})
|
||||
if err := Mount(app, cloud.Deps{Logger: luxlog.New("test")}); err != nil {
|
||||
t.Fatalf("Mount: %v", err)
|
||||
}
|
||||
return app
|
||||
}
|
||||
|
||||
// do issues a request. user simulates the SanitizeIdentity-minted X-User-Id
|
||||
// (present ONLY for a validated bearer); org simulates the minted X-Org-Id. In the
|
||||
// harness there is no middleware, so c.User()/c.Org() read these headers directly
|
||||
// — exactly the values SanitizeIdentity would set downstream.
|
||||
func do(t *testing.T, app *zip.App, method, path, user, org string) (int, []byte) {
|
||||
t.Helper()
|
||||
req := httptest.NewRequest(method, path, nil)
|
||||
if user != "" {
|
||||
req.Header.Set("X-User-Id", user)
|
||||
}
|
||||
if org != "" {
|
||||
req.Header.Set("X-Org-Id", org)
|
||||
}
|
||||
resp, err := app.Fiber().Test(req)
|
||||
if err != nil {
|
||||
t.Fatalf("Test %s %s: %v", method, path, err)
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
b, _ := io.ReadAll(resp.Body)
|
||||
return resp.StatusCode, b
|
||||
}
|
||||
|
||||
var dataEndpoints = []string{
|
||||
"/v1/analytics/overview",
|
||||
"/v1/analytics/timeseries",
|
||||
"/v1/analytics/top",
|
||||
}
|
||||
|
||||
// TestNoPrincipalForbidden: no validated principal (no X-User-Id) → 403 on every
|
||||
// data endpoint. This is the "no-Bearer → 403" contract.
|
||||
func TestNoPrincipalForbidden(t *testing.T) {
|
||||
app := mountApp(t)
|
||||
for _, p := range dataEndpoints {
|
||||
if code, _ := do(t, app, http.MethodGet, p, "", ""); code != http.StatusForbidden {
|
||||
t.Fatalf("no-principal GET %s want 403, got %d", p, code)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestForgedOrgWithoutBearerForbidden: THE cross-tenant-forge proof. A caller that
|
||||
// reaches the pod directly with a raw `X-Org-Id: maxpower` but NO validated
|
||||
// principal (no X-User-Id) is refused 403 — it can never read maxpower's analytics
|
||||
// off the Phase-1 header-passthrough path. (SanitizeIdentity leaves X-User-Id
|
||||
// empty on that path; our tenant() gate rejects it.)
|
||||
func TestForgedOrgWithoutBearerForbidden(t *testing.T) {
|
||||
app := mountApp(t)
|
||||
for _, p := range dataEndpoints {
|
||||
if code, _ := do(t, app, http.MethodGet, p, "", "maxpower"); code != http.StatusForbidden {
|
||||
t.Fatalf("forged-org-no-bearer GET %s want 403, got %d", p, code)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestDatastoreDisabledHonest503: a VALIDATED principal, but the ClickHouse ledger
|
||||
// is not connected (DatastoreEnabled()==false in this harness) → honest 503, never
|
||||
// a fake 200 with zeros. Proves the "no fabricated metrics" invariant.
|
||||
func TestDatastoreDisabledHonest503(t *testing.T) {
|
||||
app := mountApp(t)
|
||||
for _, p := range dataEndpoints {
|
||||
code, body := do(t, app, http.MethodGet, p, "user-dave", "maxpower")
|
||||
if code != http.StatusServiceUnavailable {
|
||||
t.Fatalf("datastore-down GET %s want 503, got %d (%s)", p, code, body)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestBadRangeIs400: a validated principal with an unknown ?range → 400 (before
|
||||
// the datastore is even consulted).
|
||||
func TestBadRangeIs400(t *testing.T) {
|
||||
app := mountApp(t)
|
||||
code, _ := do(t, app, http.MethodGet, "/v1/analytics/overview?range=bogus", "user-dave", "maxpower")
|
||||
if code != http.StatusBadRequest {
|
||||
t.Fatalf("bad range want 400, got %d", code)
|
||||
}
|
||||
}
|
||||
|
||||
// TestHealthOwnedByAnalyticsHonest: /v1/analytics/health is the analytics
|
||||
// subsystem's REAL probe (service=analytics, datastore bool), NOT serve.go's
|
||||
// generic GET /v1/<name>/health fake-200 (which never mounts here because we
|
||||
// register as "analyticssvc"). With the datastore down it 503s honestly. Health
|
||||
// needs no principal — liveness must be probe-able.
|
||||
func TestHealthOwnedByAnalyticsHonest(t *testing.T) {
|
||||
app := mountApp(t)
|
||||
code, body := do(t, app, http.MethodGet, "/v1/analytics/health", "", "")
|
||||
if code != http.StatusServiceUnavailable {
|
||||
t.Fatalf("health (datastore down) want 503, got %d (%s)", code, body)
|
||||
}
|
||||
var h map[string]any
|
||||
if err := json.Unmarshal(body, &h); err != nil {
|
||||
t.Fatalf("health json: %v (%s)", err, body)
|
||||
}
|
||||
if h["service"] != "analytics" {
|
||||
t.Fatalf("health service want analytics (not generic liveness), got %v", h["service"])
|
||||
}
|
||||
if h["datastore"] != false {
|
||||
t.Fatalf("health datastore want false when disconnected, got %v", h["datastore"])
|
||||
}
|
||||
if h["status"] != "degraded" {
|
||||
t.Fatalf("health status want degraded when datastore down, got %v", h["status"])
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,431 @@
|
||||
// Copyright 2023-2026 Hanzo AI Inc. All Rights Reserved.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
// Pure core of the analytics lens: SQL predicate builders, ClickHouse value
|
||||
// coercers, and the pure assemblers that turn raw ClickHouse rows into the
|
||||
// response structs. Everything here is I/O-free so the tests drive it with mock
|
||||
// rows — no ClickHouse needed — exactly as ai/object/cloud_usage.go proves out
|
||||
// its Overview assembler. The handlers (analytics.go) are the thin orchestration
|
||||
// that fetches the rows and calls these.
|
||||
//
|
||||
// THE ONE TENANCY INVARIANT lives here: llmWhere / eventsWhere ALWAYS emit
|
||||
// "… = ?" with the org bound POSITIONALLY (never interpolated), so no query this
|
||||
// package builds can read a tenant other than the caller's, and a hostile org
|
||||
// slug can never escape into SQL. The isolation test asserts this directly.
|
||||
package analytics
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"math"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Warehouse + tables (the ONE analytics warehouse per unified-analytics.md §1).
|
||||
const (
|
||||
llmTable = "hanzo.cloud_usage" // live LLM usage ledger (real data today)
|
||||
eventsTable = "hanzo.events" // web/commerce/UI wide event table (honest-empty until the collector emits)
|
||||
)
|
||||
|
||||
// ── Tenancy predicates (the isolation boundary) ─────────────────────────────
|
||||
//
|
||||
// Both builders bind the org POSITIONALLY. The time bounds are bound too (as
|
||||
// ClickHouse DateTime string literals, the proven cloud_usage.go transport), so
|
||||
// NOTHING user-derived is ever interpolated. cloud_usage keys the tenant on
|
||||
// `organization`; hanzo.events keys it on `tenant_id` (== the IAM org slug).
|
||||
|
||||
// llmWhere is the org-scoped time predicate for hanzo.cloud_usage. org is the
|
||||
// validated IAM owner slug, passed EXACTLY (the ledger stored it verbatim); it is
|
||||
// always the trailing bound parameter.
|
||||
func llmWhere(org string, start, end time.Time) (string, []any) {
|
||||
return "timestamp >= ? AND timestamp < ? AND organization = ?",
|
||||
[]any{tsLiteral(start), tsLiteral(end), org}
|
||||
}
|
||||
|
||||
// eventsWhere is the org-scoped time predicate for hanzo.events. Same shape as
|
||||
// llmWhere but keyed on `tenant_id` (the events table's canonical org column).
|
||||
func eventsWhere(org string, start, end time.Time) (string, []any) {
|
||||
return "timestamp >= ? AND timestamp < ? AND tenant_id = ?",
|
||||
[]any{tsLiteral(start), tsLiteral(end), org}
|
||||
}
|
||||
|
||||
// tsLiteral formats a time as a ClickHouse DateTime literal (UTC). Bound as a
|
||||
// string arg — identical to ai/object/cloud_usage.go's cloudUsageTS.
|
||||
func tsLiteral(t time.Time) string { return t.UTC().Format("2006-01-02 15:04:05") }
|
||||
|
||||
// ── Response types ──────────────────────────────────────────────────────────
|
||||
|
||||
type Scope struct {
|
||||
Org string `json:"org"`
|
||||
}
|
||||
|
||||
// LLMOverview is the flagship lens: real per-org KPIs from hanzo.cloud_usage.
|
||||
type LLMOverview struct {
|
||||
Available bool `json:"available"`
|
||||
Requests int64 `json:"requests"`
|
||||
Tokens int64 `json:"tokens"`
|
||||
PromptTokens int64 `json:"promptTokens"`
|
||||
CompletionTokens int64 `json:"completionTokens"`
|
||||
SpendCents int64 `json:"spendCents"`
|
||||
Models int64 `json:"models"`
|
||||
Providers int64 `json:"providers"`
|
||||
Errors int64 `json:"errors"`
|
||||
ErrorRate float64 `json:"errorRate"` // 0..1, errors/requests
|
||||
Source string `json:"source"`
|
||||
}
|
||||
|
||||
// WebOverview is the web lens over hanzo.events. Honest-empty (Available=false)
|
||||
// until the collector emits web events.
|
||||
type WebOverview struct {
|
||||
Available bool `json:"available"`
|
||||
Reason string `json:"reason,omitempty"`
|
||||
Pageviews int64 `json:"pageviews"`
|
||||
Visitors int64 `json:"visitors"`
|
||||
Sessions int64 `json:"sessions"`
|
||||
Source string `json:"source"`
|
||||
}
|
||||
|
||||
// CommerceOverview is the commerce lens over hanzo.events. Honest-empty until
|
||||
// commerce emits order events.
|
||||
type CommerceOverview struct {
|
||||
Available bool `json:"available"`
|
||||
Reason string `json:"reason,omitempty"`
|
||||
Orders int64 `json:"orders"`
|
||||
Revenue float64 `json:"revenue"`
|
||||
AOV float64 `json:"aov"` // revenue/orders
|
||||
Source string `json:"source"`
|
||||
}
|
||||
|
||||
type Overview struct {
|
||||
Range string `json:"range"`
|
||||
Start string `json:"start"`
|
||||
End string `json:"end"`
|
||||
Interval string `json:"interval"`
|
||||
Scope Scope `json:"scope"`
|
||||
LLM LLMOverview `json:"llm"`
|
||||
Web WebOverview `json:"web"`
|
||||
Commerce CommerceOverview `json:"commerce"`
|
||||
}
|
||||
|
||||
type SeriesPoint struct {
|
||||
T string `json:"t"` // RFC3339 bucket start (UTC)
|
||||
Requests int64 `json:"requests"`
|
||||
Tokens int64 `json:"tokens"`
|
||||
SpendCents int64 `json:"spendCents"`
|
||||
}
|
||||
|
||||
type Timeseries struct {
|
||||
Range string `json:"range"`
|
||||
Start string `json:"start"`
|
||||
End string `json:"end"`
|
||||
Interval string `json:"interval"`
|
||||
Scope Scope `json:"scope"`
|
||||
Series []SeriesPoint `json:"series"`
|
||||
Source string `json:"source"`
|
||||
}
|
||||
|
||||
type ModelRow struct {
|
||||
Model string `json:"model"`
|
||||
Provider string `json:"provider"`
|
||||
Requests int64 `json:"requests"`
|
||||
Tokens int64 `json:"tokens"`
|
||||
SpendCents int64 `json:"spendCents"`
|
||||
Pct float64 `json:"pct"` // share of total spend, 0..100
|
||||
}
|
||||
|
||||
type TopModels struct {
|
||||
Available bool `json:"available"`
|
||||
Items []ModelRow `json:"items"`
|
||||
Source string `json:"source"`
|
||||
}
|
||||
|
||||
type ProductRow struct {
|
||||
ProductID string `json:"productId"`
|
||||
Orders int64 `json:"orders"`
|
||||
Revenue float64 `json:"revenue"`
|
||||
Units int64 `json:"units"`
|
||||
}
|
||||
|
||||
type TopProducts struct {
|
||||
Available bool `json:"available"`
|
||||
Reason string `json:"reason,omitempty"`
|
||||
Items []ProductRow `json:"items"`
|
||||
Source string `json:"source"`
|
||||
}
|
||||
|
||||
type Top struct {
|
||||
Range string `json:"range"`
|
||||
Start string `json:"start"`
|
||||
End string `json:"end"`
|
||||
Scope Scope `json:"scope"`
|
||||
Models TopModels `json:"models"`
|
||||
Products TopProducts `json:"products"`
|
||||
}
|
||||
|
||||
// ── Pure assemblers ─────────────────────────────────────────────────────────
|
||||
|
||||
// buildLLMOverview assembles the LLM KPI block from the single aggregate row.
|
||||
// A nil/empty row yields honest zeros (Available is still true — the datastore
|
||||
// answered; there is simply no usage in the window). Pure.
|
||||
func buildLLMOverview(row map[string]any) LLMOverview {
|
||||
requests := aInt64(row["requests"])
|
||||
errors := aInt64(row["errors"])
|
||||
o := LLMOverview{
|
||||
Available: true,
|
||||
Requests: requests,
|
||||
Tokens: aInt64(row["tokens"]),
|
||||
PromptTokens: aInt64(row["prompt_tokens"]),
|
||||
CompletionTokens: aInt64(row["completion_tokens"]),
|
||||
SpendCents: aInt64(row["cost_cents"]),
|
||||
Models: aInt64(row["models"]),
|
||||
Providers: aInt64(row["providers"]),
|
||||
Errors: errors,
|
||||
Source: llmTable,
|
||||
}
|
||||
if requests > 0 {
|
||||
o.ErrorRate = round3(float64(errors) / float64(requests))
|
||||
}
|
||||
return o
|
||||
}
|
||||
|
||||
// buildWebOverview / buildCommerceOverview assemble the events lenses. The
|
||||
// handler passes ok=false when the events query failed (table absent) so the
|
||||
// lens is honestly reported unavailable rather than as fabricated zeros.
|
||||
func buildWebOverview(row map[string]any, ok bool) WebOverview {
|
||||
w := WebOverview{Available: ok, Source: eventsTable}
|
||||
if !ok {
|
||||
w.Reason = "no web analytics events yet"
|
||||
return w
|
||||
}
|
||||
w.Pageviews = aInt64(row["pageviews"])
|
||||
w.Visitors = aInt64(row["visitors"])
|
||||
w.Sessions = aInt64(row["sessions"])
|
||||
return w
|
||||
}
|
||||
|
||||
func buildCommerceOverview(row map[string]any, ok bool) CommerceOverview {
|
||||
c := CommerceOverview{Available: ok, Source: eventsTable}
|
||||
if !ok {
|
||||
c.Reason = "no commerce events yet"
|
||||
return c
|
||||
}
|
||||
c.Orders = aInt64(row["orders"])
|
||||
c.Revenue = aFloat64(row["revenue"])
|
||||
if c.Orders > 0 {
|
||||
c.AOV = round2(c.Revenue / float64(c.Orders))
|
||||
}
|
||||
return c
|
||||
}
|
||||
|
||||
// buildSeries turns sparse ClickHouse buckets into an evenly-spaced, gap-filled
|
||||
// series so the client charts a continuous line. Bucket alignment matches
|
||||
// toStartOf{Hour,Day}(…, 'UTC'): Go's Truncate over the step lands on the same
|
||||
// UTC boundaries. Pure (mirrors ai/object buildCloudUsageSeries).
|
||||
func buildSeries(start, end time.Time, interval string, rows []map[string]any) []SeriesPoint {
|
||||
step := stepOf(interval)
|
||||
|
||||
type agg struct{ requests, tokens, spend int64 }
|
||||
idx := make(map[int64]agg, len(rows))
|
||||
for _, r := range rows {
|
||||
bt := aTime(r["bucket"]).Truncate(step)
|
||||
idx[bt.Unix()] = agg{
|
||||
requests: aInt64(r["requests"]),
|
||||
tokens: aInt64(r["tokens"]),
|
||||
spend: aInt64(r["cost_cents"]),
|
||||
}
|
||||
}
|
||||
|
||||
out := make([]SeriesPoint, 0, 64)
|
||||
for t := start.UTC().Truncate(step); t.Before(end); t = t.Add(step) {
|
||||
a := idx[t.Unix()]
|
||||
out = append(out, SeriesPoint{
|
||||
T: t.UTC().Format(time.RFC3339),
|
||||
Requests: a.requests,
|
||||
Tokens: a.tokens,
|
||||
SpendCents: a.spend,
|
||||
})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func stepOf(interval string) time.Duration {
|
||||
if strings.EqualFold(interval, "day") {
|
||||
return 24 * time.Hour
|
||||
}
|
||||
return time.Hour
|
||||
}
|
||||
|
||||
// buildTopModels assembles the top-models table, computing each model's share of
|
||||
// total spend. Rows arrive already ordered by the query, but we sort defensively
|
||||
// so the pct/ordering is correct regardless of driver row order. Pure.
|
||||
func buildTopModels(rows []map[string]any) TopModels {
|
||||
items := make([]ModelRow, 0, len(rows))
|
||||
var totalCents int64
|
||||
for _, r := range rows {
|
||||
spend := aInt64(r["cost_cents"])
|
||||
totalCents += spend
|
||||
items = append(items, ModelRow{
|
||||
Model: aString(r["model"]),
|
||||
Provider: aString(r["provider"]),
|
||||
Requests: aInt64(r["requests"]),
|
||||
Tokens: aInt64(r["tokens"]),
|
||||
SpendCents: spend,
|
||||
})
|
||||
}
|
||||
sort.SliceStable(items, func(i, j int) bool {
|
||||
if items[i].SpendCents != items[j].SpendCents {
|
||||
return items[i].SpendCents > items[j].SpendCents
|
||||
}
|
||||
return items[i].Requests > items[j].Requests
|
||||
})
|
||||
for i := range items {
|
||||
items[i].Pct = pctOf(items[i].SpendCents, totalCents)
|
||||
}
|
||||
return TopModels{Available: true, Items: items, Source: llmTable}
|
||||
}
|
||||
|
||||
// buildTopProducts assembles the top-products table from hanzo.events. ok=false
|
||||
// (events table absent) → honest-empty. Pure.
|
||||
func buildTopProducts(rows []map[string]any, ok bool) TopProducts {
|
||||
if !ok {
|
||||
return TopProducts{Available: false, Reason: "no commerce events yet", Items: []ProductRow{}, Source: eventsTable}
|
||||
}
|
||||
items := make([]ProductRow, 0, len(rows))
|
||||
for _, r := range rows {
|
||||
items = append(items, ProductRow{
|
||||
ProductID: aString(r["productId"]),
|
||||
Orders: aInt64(r["orders"]),
|
||||
Revenue: aFloat64(r["revenue"]),
|
||||
Units: aInt64(r["units"]),
|
||||
})
|
||||
}
|
||||
return TopProducts{Available: true, Items: items, Source: eventsTable}
|
||||
}
|
||||
|
||||
// ── Value coercion ──────────────────────────────────────────────────────────
|
||||
//
|
||||
// The direct ClickHouse driver decodes each column to its native Go scan type
|
||||
// (uint64 for count()/sum(UInt*), float64 for toFloat64, time.Time for DateTime,
|
||||
// string for String). These coercers accept those natives AND the JSON-transport
|
||||
// fallbacks (float64/json.Number/string) so a transport change can't crash a read.
|
||||
|
||||
func aInt64(v any) int64 {
|
||||
switch n := v.(type) {
|
||||
case nil:
|
||||
return 0
|
||||
case int:
|
||||
return int64(n)
|
||||
case int64:
|
||||
return n
|
||||
case int32:
|
||||
return int64(n)
|
||||
case uint:
|
||||
return int64(n)
|
||||
case uint64:
|
||||
return int64(n)
|
||||
case uint32:
|
||||
return int64(n)
|
||||
case uint16:
|
||||
return int64(n)
|
||||
case uint8:
|
||||
return int64(n)
|
||||
case float64:
|
||||
return int64(n)
|
||||
case float32:
|
||||
return int64(n)
|
||||
case json.Number:
|
||||
i, _ := n.Int64()
|
||||
return i
|
||||
case string:
|
||||
if i, err := strconv.ParseInt(strings.TrimSpace(n), 10, 64); err == nil {
|
||||
return i
|
||||
}
|
||||
if f, err := strconv.ParseFloat(strings.TrimSpace(n), 64); err == nil {
|
||||
return int64(f)
|
||||
}
|
||||
return 0
|
||||
default:
|
||||
return 0
|
||||
}
|
||||
}
|
||||
|
||||
func aFloat64(v any) float64 {
|
||||
switch n := v.(type) {
|
||||
case nil:
|
||||
return 0
|
||||
case float64:
|
||||
return n
|
||||
case float32:
|
||||
return float64(n)
|
||||
case int:
|
||||
return float64(n)
|
||||
case int64:
|
||||
return float64(n)
|
||||
case uint64:
|
||||
return float64(n)
|
||||
case json.Number:
|
||||
f, _ := n.Float64()
|
||||
return f
|
||||
case string:
|
||||
f, _ := strconv.ParseFloat(strings.TrimSpace(n), 64)
|
||||
return f
|
||||
default:
|
||||
return 0
|
||||
}
|
||||
}
|
||||
|
||||
func aString(v any) string {
|
||||
switch s := v.(type) {
|
||||
case nil:
|
||||
return ""
|
||||
case string:
|
||||
return s
|
||||
case fmt.Stringer:
|
||||
return s.String()
|
||||
default:
|
||||
return fmt.Sprintf("%v", s)
|
||||
}
|
||||
}
|
||||
|
||||
func aTime(v any) time.Time {
|
||||
if t, ok := v.(time.Time); ok {
|
||||
return t.UTC()
|
||||
}
|
||||
s := strings.TrimSpace(aString(v))
|
||||
if s != "" {
|
||||
for _, layout := range []string{"2006-01-02 15:04:05", time.RFC3339, "2006-01-02"} {
|
||||
if t, err := time.Parse(layout, s); err == nil {
|
||||
return t.UTC()
|
||||
}
|
||||
}
|
||||
}
|
||||
if n := aInt64(v); n > 0 {
|
||||
return time.Unix(n, 0).UTC()
|
||||
}
|
||||
return time.Time{}
|
||||
}
|
||||
|
||||
func pctOf(part, total int64) float64 {
|
||||
if total <= 0 {
|
||||
return 0
|
||||
}
|
||||
return round1(float64(part) / float64(total) * 100)
|
||||
}
|
||||
|
||||
func round1(f float64) float64 { return math.Round(f*10) / 10 }
|
||||
func round2(f float64) float64 { return math.Round(f*100) / 100 }
|
||||
func round3(f float64) float64 { return math.Round(f*1000) / 1000 }
|
||||
@@ -91,6 +91,7 @@ import (
|
||||
// clients/prompts is the red-approved, versioned prompt library and the ONE
|
||||
// owner of /v1/prompts/* (it supersedes the earlier clients/prompt facade).
|
||||
_ "github.com/hanzoai/cloud/clients/agents" // order 127 — /v1/agents/*
|
||||
_ "github.com/hanzoai/cloud/clients/analytics" // order 132 — /v1/analytics/* (native-Go analytics on datastore/ClickHouse: per-org LLM usage + web/commerce lenses)
|
||||
_ "github.com/hanzoai/cloud/clients/crm" // order 131 — /v1/crm/* (native-Go CRM on Base: companies/contacts/opportunities)
|
||||
_ "github.com/hanzoai/cloud/clients/functions" // order 128 — /v1/functions/*
|
||||
_ "github.com/hanzoai/cloud/clients/git" // order 132 — /v1/git/* (S3-backed native Git hosting; smart-HTTP clone/push)
|
||||
|
||||
Reference in New Issue
Block a user