diff --git a/internal/app/app_test.go b/internal/app/app_test.go index 982ffff..8a010bd 100644 --- a/internal/app/app_test.go +++ b/internal/app/app_test.go @@ -162,8 +162,8 @@ func (f *fakeAnalytics) Query(context.Context, analytics.QueryRequest, agent.Tok return analytics.QueryResponse{}, nil } -func (f *fakeAnalytics) Schema(context.Context, agent.Token) ([]analytics.QueryTable, error) { - return nil, nil +func (f *fakeAnalytics) Schema(context.Context, agent.Token) (analytics.SchemaResponse, error) { + return analytics.SchemaResponse{}, nil } func TestBuildConsentStrategyAgentConfirmsIcons(t *testing.T) { diff --git a/internal/cli/analytics.go b/internal/cli/analytics.go index 965c3a2..3891766 100644 --- a/internal/cli/analytics.go +++ b/internal/cli/analytics.go @@ -1,6 +1,7 @@ package cli import ( + "errors" "fmt" "strings" @@ -8,6 +9,7 @@ import ( "github.com/tollbit/cli/internal/app" analyticsclient "github.com/tollbit/cli/internal/client/analytics" "github.com/tollbit/cli/internal/credentials/agenttoken" + "github.com/tollbit/cli/internal/errorsx/problemjson" ) const analyticsLongHelp = `Query TollBit analytics for your organization's sites. @@ -31,13 +33,19 @@ unfiltered queries are rejected even with LIMIT. The server also caps the number of rows returned, so add ORDER BY with LIMIT and OFFSET when paging through large results. -Output is a JSON object with "columns" (name and type) and "rows" (arrays in -column order, null for missing values).` +Output is a JSON object with "columns" (name and type), "rows" (arrays in +column order, null for missing values), and "meta" (row_count, truncated, +bytes_scanned, duration_ms). When meta.truncated is true the result was cut +at the server's row limit and a warning is printed to stderr.` const analyticsSchemaLongHelp = `List the analytics tables and columns available to your organization. -Output is a JSON array of tables, each with its name and columns (name and -type). Run this before "analytics query" to discover table and column names.` +Output is a JSON object with "dialect", "tables" and "limits". Each table has +its name, a description, its columns (name, type, description and, where the +column has a fixed set, values), and "clustering": the columns the table is +ordered by, most significant first. Filtering and grouping in that order, +after a timestamp filter, scans the least data. "limits" maps each server +limit to its value, unit and description. Run this before "analytics query".` const analyticsQueryExample = ` # Discover tables and columns first tollbit analytics schema @@ -116,14 +124,36 @@ func runAnalyticsQuery(cmd *cobra.Command, factory app.Factory, sql string) erro } result, err := analyticsClient.Query(cmd.Context(), analyticsclient.QueryRequest{SQL: sql}, token) if err != nil { + if hint := analyticsErrorHint(err); hint != "" { + printLeadingCommand(cmd.ErrOrStderr(), hint) + } return RuntimeError(fmt.Errorf("error querying analytics: %w", err)) } if err := writeJSON(cmd.OutOrStdout(), result); err != nil { return RuntimeError(fmt.Errorf("error writing analytics response: %w", err)) } + if result.Meta != nil && result.Meta.Truncated { + printLeadingCommand(cmd.ErrOrStderr(), fmt.Sprintf("warning: result truncated at %d rows (server limit). Add ORDER BY with LIMIT and OFFSET to page, or narrow the query.", result.Meta.RowCount)) + } return nil } +// analyticsErrorHint maps a server error code to a next step. Unknown or +// absent codes give no hint. +func analyticsErrorHint(err error) string { + var problem problemjson.Problem + if !errors.As(err, &problem) || problem.Code == nil { + return "" + } + switch string(*problem.Code) { + case "analytics_unknown_table", "analytics_statement_not_allowed": + return "Run \"tollbit analytics schema\" to list the available tables." + case "analytics_scan_limit_exceeded", "analytics_query_timeout": + return "Run \"tollbit analytics schema\" to see the query limits, then filter on timestamp or select fewer columns." + } + return "" +} + func NewAnalyticsSchemaCommand(factory app.Factory) *cobra.Command { cmd := &cobra.Command{ Use: "schema", diff --git a/internal/cli/analytics_test.go b/internal/cli/analytics_test.go index 98f7404..9ddfac0 100644 --- a/internal/cli/analytics_test.go +++ b/internal/cli/analytics_test.go @@ -47,7 +47,7 @@ func TestAnalyticsQueryUsesOBOAgentTokenAndWritesJSON(t *testing.T) { if request.SQL != "SELECT * FROM logs" { t.Fatalf("unexpected SQL: %q", request.SQL) } - _, _ = w.Write([]byte(`{"columns":[{"name":"requests","type":"INTEGER"},{"name":"optional","type":"STRING"}],"rows":[[42,null]]}`)) + _, _ = w.Write([]byte(`{"columns":[{"name":"requests","type":"INTEGER"},{"name":"optional","type":"STRING"}],"rows":[[42,null]],"meta":{"row_count":1,"truncated":false,"bytes_scanned":10,"duration_ms":5}}`)) })) defer analyticsSrv.Close() @@ -68,6 +68,10 @@ func TestAnalyticsQueryUsesOBOAgentTokenAndWritesJSON(t *testing.T) { Type string `json:"type"` } `json:"columns"` Rows [][]any `json:"rows"` + Meta *struct { + RowCount int `json:"row_count"` + Truncated bool `json:"truncated"` + } `json:"meta"` } if err := json.Unmarshal(stdout.Bytes(), &output); err != nil { t.Fatalf("invalid JSON output %q: %v", stdout.String(), err) @@ -78,6 +82,87 @@ func TestAnalyticsQueryUsesOBOAgentTokenAndWritesJSON(t *testing.T) { if len(output.Rows) != 1 || output.Rows[0][0] != float64(42) || output.Rows[0][1] != nil { t.Fatalf("unexpected rows: %#v", output.Rows) } + if output.Meta == nil || output.Meta.RowCount != 1 { + t.Fatalf("expected meta in output, got %q", stdout.String()) + } + if strings.Contains(stderr.String(), "truncated") { + t.Fatalf("no truncation warning expected, got %q", stderr.String()) + } +} + +func TestAnalyticsQueryWarnsOnTruncation(t *testing.T) { + token := testAgentJWTWithOBO(t) + storageDir := t.TempDir() + if err := os.WriteFile(filepath.Join(storageDir, "agent-token.jwt"), []byte(token), 0o600); err != nil { + t.Fatal(err) + } + analyticsSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(`{"columns":[{"name":"n","type":"INT64"}],"rows":[[1]],"meta":{"row_count":10000,"truncated":true,"bytes_scanned":10,"duration_ms":5}}`)) + })) + defer analyticsSrv.Close() + + config := testConfig() + config.Analytics.Enabled = true + config.Analytics.BaseURL = analyticsSrv.URL + config.Credentials.StorageDir = storageDir + config.Runtime.StateDir = storageDir + + var stdout, stderr bytes.Buffer + code := executeTestCommandWithConfig(config, []string{"analytics", "query", "SELECT 1"}, nil, &stdout, &stderr) + if code != 0 { + t.Fatalf("expected success, got %d (stderr=%q)", code, stderr.String()) + } + if !strings.Contains(stderr.String(), "warning: result truncated at 10000 rows") { + t.Fatalf("expected truncation warning on stderr, got %q", stderr.String()) + } + if strings.Contains(stdout.String(), "warning") { + t.Fatalf("stdout must stay data-only, got %q", stdout.String()) + } +} + +func TestAnalyticsQueryPrintsHintForKnownErrorCodes(t *testing.T) { + token := testAgentJWTWithOBO(t) + storageDir := t.TempDir() + if err := os.WriteFile(filepath.Join(storageDir, "agent-token.jwt"), []byte(token), 0o600); err != nil { + t.Fatal(err) + } + for _, tc := range []struct { + body string + wantHint string + }{ + {`{"title":"Bad Request","status":400,"detail":"Unknown table. Available tables: a, b.","code":"analytics_unknown_table"}`, "tollbit analytics schema"}, + {`{"title":"Unprocessable","status":422,"detail":"Query would scan about 150 GiB, more than the limit of 100 GiB.","code":"analytics_scan_limit_exceeded"}`, "filter on timestamp"}, + {`{"title":"Bad Request","status":400,"detail":"Syntax error","code":"analytics_invalid_query"}`, ""}, + {`{"title":"Bad Request","status":400,"detail":"no code at all"}`, ""}, + } { + body := tc.body + analyticsSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/problem+json") + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte(body)) + })) + config := testConfig() + config.Analytics.Enabled = true + config.Analytics.BaseURL = analyticsSrv.URL + config.Credentials.StorageDir = storageDir + config.Runtime.StateDir = storageDir + + var stdout, stderr bytes.Buffer + code := executeTestCommandWithConfig(config, []string{"analytics", "query", "SELECT 1"}, nil, &stdout, &stderr) + analyticsSrv.Close() + if code == 0 { + t.Fatalf("expected failure for %s", body) + } + if tc.wantHint == "" { + if strings.Contains(stderr.String(), "tollbit analytics schema") { + t.Fatalf("unexpected hint for %s: %q", body, stderr.String()) + } + continue + } + if !strings.Contains(stderr.String(), tc.wantHint) { + t.Fatalf("expected hint %q for %s, got %q", tc.wantHint, body, stderr.String()) + } + } } func TestAnalyticsSchemaUsesOBOAgentTokenAndWritesJSON(t *testing.T) { @@ -93,6 +178,57 @@ func TestAnalyticsSchemaUsesOBOAgentTokenAndWritesJSON(t *testing.T) { if r.Header.Get("Authorization") != "Bearer "+token { t.Fatal("unexpected authorization header") } + _, _ = w.Write([]byte(`{"dialect":"bigquery","tables":[{"name":"agent_logs_by_page","clustering":["host","user_agent","path"],"columns":[{"name":"host","type":"STRING"}]}],"limits":{"max_rows":{"value":10000,"unit":"rows"}}}`)) + })) + defer analyticsSrv.Close() + + config := testConfig() + config.Analytics.Enabled = true + config.Analytics.BaseURL = analyticsSrv.URL + config.Credentials.StorageDir = storageDir + config.Runtime.StateDir = storageDir + + var stdout, stderr bytes.Buffer + code := executeTestCommandWithConfig(config, []string{"analytics", "schema"}, nil, &stdout, &stderr) + if code != 0 { + t.Fatalf("expected success, got %d (stderr=%q)", code, stderr.String()) + } + var output struct { + Dialect string `json:"dialect"` + Tables []struct { + Name string `json:"name"` + Clustering []string `json:"clustering"` + Columns []struct { + Name string `json:"name"` + Type string `json:"type"` + } `json:"columns"` + } `json:"tables"` + Limits map[string]struct { + Value float64 `json:"value"` + Unit string `json:"unit"` + } `json:"limits"` + } + if err := json.Unmarshal(stdout.Bytes(), &output); err != nil { + t.Fatalf("invalid JSON output %q: %v", stdout.String(), err) + } + if output.Dialect != "bigquery" || len(output.Tables) != 1 || output.Tables[0].Name != "agent_logs_by_page" { + t.Fatalf("unexpected schema: %#v", output) + } + if len(output.Tables[0].Clustering) != 3 || len(output.Tables[0].Columns) != 1 || output.Tables[0].Columns[0].Name != "host" { + t.Fatalf("unexpected table: %#v", output.Tables[0]) + } + if output.Limits["max_rows"].Value != 10000 || output.Limits["max_rows"].Unit != "rows" { + t.Fatalf("unexpected limits: %#v", output.Limits) + } +} + +func TestAnalyticsSchemaAcceptsLegacyArrayAndPrintsObject(t *testing.T) { + token := testAgentJWTWithOBO(t) + storageDir := t.TempDir() + if err := os.WriteFile(filepath.Join(storageDir, "agent-token.jwt"), []byte(token), 0o600); err != nil { + t.Fatal(err) + } + analyticsSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { _, _ = w.Write([]byte(`[{"name":"agent_logs_by_page","columns":[{"name":"host","type":"STRING"}]}]`)) })) defer analyticsSrv.Close() @@ -108,21 +244,16 @@ func TestAnalyticsSchemaUsesOBOAgentTokenAndWritesJSON(t *testing.T) { if code != 0 { t.Fatalf("expected success, got %d (stderr=%q)", code, stderr.String()) } - var output []struct { - Name string `json:"name"` - Columns []struct { + var output struct { + Tables []struct { Name string `json:"name"` - Type string `json:"type"` - } `json:"columns"` + } `json:"tables"` } if err := json.Unmarshal(stdout.Bytes(), &output); err != nil { t.Fatalf("invalid JSON output %q: %v", stdout.String(), err) } - if len(output) != 1 || output[0].Name != "agent_logs_by_page" { - t.Fatalf("unexpected tables: %#v", output) - } - if len(output[0].Columns) != 1 || output[0].Columns[0].Name != "host" { - t.Fatalf("unexpected columns: %#v", output[0].Columns) + if len(output.Tables) != 1 || output.Tables[0].Name != "agent_logs_by_page" { + t.Fatalf("unexpected tables: %#v", output.Tables) } } @@ -173,8 +304,8 @@ func TestAnalyticsHelpDocumentsQueryContract(t *testing.T) { want []string }{ {[]string{"analytics", "--help"}, []string{"analytics schema", "analytics query"}}, - {[]string{"analytics", "query", "--help"}, []string{"BigQuery Standard SQL", "single SELECT", "timestamp", "user_agent_aggregate", "Examples:"}}, - {[]string{"analytics", "schema", "--help"}, []string{"JSON array of tables", "analytics query"}}, + {[]string{"analytics", "query", "--help"}, []string{"BigQuery Standard SQL", "single SELECT", "timestamp", "user_agent_aggregate", "Examples:", "meta"}}, + {[]string{"analytics", "schema", "--help"}, []string{"JSON object", "limits", "clustering", "analytics query"}}, } { var stdout, stderr bytes.Buffer code := executeTestCommandWithConfig(config, tc.args, nil, &stdout, &stderr) diff --git a/internal/client/analytics/client.go b/internal/client/analytics/client.go index d8fd965..58c9075 100644 --- a/internal/client/analytics/client.go +++ b/internal/client/analytics/client.go @@ -27,7 +27,7 @@ type ( Client interface { Query(context.Context, QueryRequest, agent.Token) (QueryResponse, error) - Schema(context.Context, agent.Token) ([]QueryTable, error) + Schema(context.Context, agent.Token) (SchemaResponse, error) } client struct { @@ -40,18 +40,46 @@ type ( } QueryColumn struct { - Name string `json:"name"` - Type string `json:"type"` + Name string `json:"name"` + Type string `json:"type"` + Description string `json:"description,omitempty"` + Values []string `json:"values,omitempty"` + } + + // QueryMeta is sent by servers that report result metadata. Absent on + // older servers, so it is a pointer and omitted when nil. + QueryMeta struct { + RowCount int `json:"row_count"` + Truncated bool `json:"truncated"` + BytesScanned int64 `json:"bytes_scanned"` + DurationMs int64 `json:"duration_ms"` } QueryResponse struct { Columns []QueryColumn `json:"columns"` Rows [][]any `json:"rows"` + Meta *QueryMeta `json:"meta,omitempty"` } QueryTable struct { - Name string `json:"name"` - Columns []QueryColumn `json:"columns"` + Name string `json:"name"` + Description string `json:"description,omitempty"` + Columns []QueryColumn `json:"columns"` + Clustering []string `json:"clustering,omitempty"` + } + + Limit struct { + Value json.Number `json:"value"` + Unit string `json:"unit"` + Description string `json:"description,omitempty"` + } + + // SchemaResponse is the schema object. Older servers return a bare array + // of tables; Schema accepts both and always returns this shape. + SchemaResponse struct { + Dialect string `json:"dialect,omitempty"` + Tables []QueryTable `json:"tables"` + Limits map[string]Limit `json:"limits,omitempty"` } ) @@ -111,36 +139,53 @@ func (c *client) Query(ctx context.Context, request QueryRequest, token agent.To return result, nil } -func (c *client) Schema(ctx context.Context, token agent.Token) ([]QueryTable, error) { +func (c *client) Schema(ctx context.Context, token agent.Token) (SchemaResponse, error) { if strings.TrimSpace(token.RawToken) == "" { - return nil, errors.New("agent token is required") + return SchemaResponse{}, errors.New("agent token is required") } if err := token.Validate(); err != nil { - return nil, err + return SchemaResponse{}, err } u := *c.baseURL u.Path = strings.TrimRight(c.baseURL.Path, "/") + schemaPath req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil) if err != nil { - return nil, err + return SchemaResponse{}, err } req.Header.Set("Accept", "application/json") req.Header.Set("Authorization", "Bearer "+token.RawToken) resp, err := c.http.Do(req) if err != nil { - return nil, err + return SchemaResponse{}, err } defer resp.Body.Close() if resp.StatusCode < 200 || resp.StatusCode >= 300 { body, _ := io.ReadAll(resp.Body) - return nil, errorsx.ParseResponseError(ctx, resp.Status, resp.StatusCode, resp.Header, body) + return SchemaResponse{}, errorsx.ParseResponseError(ctx, resp.Status, resp.StatusCode, resp.Header, body) } - var result []QueryTable - if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { - return nil, err + body, err := io.ReadAll(resp.Body) + if err != nil { + return SchemaResponse{}, err + } + return decodeSchema(body) +} + +// decodeSchema accepts the schema object or the older bare array of tables. +func decodeSchema(body []byte) (SchemaResponse, error) { + trimmed := bytes.TrimLeft(body, " \t\r\n") + if len(trimmed) > 0 && trimmed[0] == '[' { + var tables []QueryTable + if err := json.Unmarshal(trimmed, &tables); err != nil { + return SchemaResponse{}, err + } + return SchemaResponse{Tables: tables}, nil + } + var result SchemaResponse + if err := json.Unmarshal(trimmed, &result); err != nil { + return SchemaResponse{}, err } return result, nil } diff --git a/internal/client/analytics/client_test.go b/internal/client/analytics/client_test.go index 5f1294a..edb2974 100644 --- a/internal/client/analytics/client_test.go +++ b/internal/client/analytics/client_test.go @@ -34,10 +34,7 @@ func TestQuery(t *testing.T) { if request.SQL != "SELECT * FROM logs" { t.Fatalf("unexpected SQL: %q", request.SQL) } - _ = json.NewEncoder(w).Encode(QueryResponse{ - Columns: []QueryColumn{{Name: "requests", Type: "INTEGER"}, {Name: "optional", Type: "STRING"}}, - Rows: [][]any{{42, nil}}, - }) + _, _ = w.Write([]byte(`{"columns":[{"name":"requests","type":"INTEGER"},{"name":"optional","type":"STRING"}],"rows":[[42,null]],"meta":{"row_count":1,"truncated":true,"bytes_scanned":512,"duration_ms":7}}`)) })) defer srv.Close() @@ -55,6 +52,27 @@ func TestQuery(t *testing.T) { if len(response.Rows) != 1 || response.Rows[0][0] != float64(42) || response.Rows[0][1] != nil { t.Fatalf("unexpected rows: %#v", response.Rows) } + if response.Meta == nil || !response.Meta.Truncated || response.Meta.RowCount != 1 || response.Meta.BytesScanned != 512 || response.Meta.DurationMs != 7 { + t.Fatalf("unexpected meta: %#v", response.Meta) + } +} + +func TestQueryWithoutMeta(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(`{"columns":[{"name":"n","type":"INT64"}],"rows":[[1]]}`)) + })) + defer srv.Close() + client, err := NewClient(Config{BaseURL: srv.URL}) + if err != nil { + t.Fatal(err) + } + response, err := client.Query(context.Background(), QueryRequest{SQL: "SELECT 1"}, validAgentToken(t)) + if err != nil { + t.Fatal(err) + } + if response.Meta != nil { + t.Fatalf("expected no meta from an older server, got %#v", response.Meta) + } } func TestSchema(t *testing.T) { @@ -69,10 +87,14 @@ func TestSchema(t *testing.T) { if r.Header.Get("Authorization") != "Bearer "+token.RawToken { t.Fatal("unexpected authorization header") } - _ = json.NewEncoder(w).Encode([]QueryTable{{ - Name: "agent_logs_by_page", - Columns: []QueryColumn{{Name: "host", Type: "STRING"}}, - }}) + _, _ = w.Write([]byte(`{ + "dialect": "bigquery", + "tables": [{"name": "agent_logs_by_page", "description": "Daily counts.", "clustering": ["host", "user_agent", "path"], + "columns": [{"name": "host", "type": "STRING", "description": "Site hostname."}, + {"name": "type", "type": "STRING", "values": ["REQUEST", "ROBOT"]}]}], + "limits": {"max_rows": {"value": 10000, "unit": "rows", "description": "Row cap."}, + "something_new": {"value": 3, "unit": "widgets"}} + }`)) })) defer srv.Close() @@ -80,15 +102,46 @@ func TestSchema(t *testing.T) { if err != nil { t.Fatal(err) } - tables, err := client.Schema(context.Background(), token) + schema, err := client.Schema(context.Background(), token) + if err != nil { + t.Fatal(err) + } + if schema.Dialect != "bigquery" { + t.Fatalf("unexpected dialect: %q", schema.Dialect) + } + if len(schema.Tables) != 1 || schema.Tables[0].Name != "agent_logs_by_page" || schema.Tables[0].Description != "Daily counts." { + t.Fatalf("unexpected tables: %#v", schema.Tables) + } + if got := schema.Tables[0].Clustering; len(got) != 3 || got[1] != "user_agent" { + t.Fatalf("unexpected clustering: %#v", got) + } + cols := schema.Tables[0].Columns + if len(cols) != 2 || cols[0].Description != "Site hostname." || len(cols[1].Values) != 2 { + t.Fatalf("unexpected columns: %#v", cols) + } + if len(schema.Limits) != 2 || schema.Limits["max_rows"].Unit != "rows" || schema.Limits["something_new"].Value.String() != "3" { + t.Fatalf("unexpected limits: %#v", schema.Limits) + } +} + +func TestSchemaAcceptsLegacyArray(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(` [{"name":"agent_logs_by_page","columns":[{"name":"host","type":"STRING"}]}]`)) + })) + defer srv.Close() + client, err := NewClient(Config{BaseURL: srv.URL}) + if err != nil { + t.Fatal(err) + } + schema, err := client.Schema(context.Background(), validAgentToken(t)) if err != nil { t.Fatal(err) } - if len(tables) != 1 || tables[0].Name != "agent_logs_by_page" { - t.Fatalf("unexpected tables: %#v", tables) + if schema.Dialect != "" || schema.Limits != nil { + t.Fatalf("legacy array must not invent dialect or limits: %#v", schema) } - if len(tables[0].Columns) != 1 || tables[0].Columns[0].Name != "host" { - t.Fatalf("unexpected columns: %#v", tables[0].Columns) + if len(schema.Tables) != 1 || schema.Tables[0].Name != "agent_logs_by_page" || schema.Tables[0].Columns[0].Name != "host" { + t.Fatalf("unexpected tables: %#v", schema.Tables) } }