wavehouse

package module
v0.0.0-...-51d8b45 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 11, 2026 License: Apache-2.0 Imports: 18 Imported by: 0

README

WaveHouse Go SDK

Official Go client for WaveHouse — a schema-aware real-time API gateway for ClickHouse.

Zero third-party runtime dependencies — stdlib only.

Full SDK documentation on wavehouse.dev

Install

go get github.com/Wave-RF/WaveHouse/clients/go

Quick Start

package main

import (
    "context"
    "fmt"
    "log"

    wavehouse "github.com/Wave-RF/WaveHouse/clients/go"
)

func main() {
    // Create an unauthenticated client (uses the server's default_role).
    client := wavehouse.NewClient(wavehouse.Config{
        BaseURL: "http://localhost:8080",
    })

    // Health check.
    if err := client.Sys.Health(context.Background()); err != nil {
        log.Fatal(err)
    }

    ctx := context.Background()

    // Insert a row.
    _, err := client.From("clicks").Insert(ctx, map[string]any{
        "page": "/home", "button": "cta",
    })
    if err != nil {
        log.Fatal(err)
    }

    // Query with the fluent builder.
    page, err := client.From("clicks").
        Select("page", "button").
        Where("page", wavehouse.OpEq, "/home").
        OrderBy("page", "asc").
        Limit(10).
        FetchUntyped(ctx)
    if err != nil {
        log.Fatal(err)
    }
    for _, row := range page.Data {
        fmt.Println(row["page"], row["button"])
    }
}

Authentication

// Static token.
client := wavehouse.NewClient(wavehouse.Config{
    BaseURL: "http://localhost:8080",
    Auth:    wavehouse.StaticToken("your-jwt"),
})

// Dynamic token (e.g. rotated).
client = wavehouse.NewClient(wavehouse.Config{
    BaseURL: "http://localhost:8080",
    Auth: func(ctx context.Context) (string, error) {
        return fetchFreshToken(ctx)
    },
})

Typed Queries (Generics)

type ClickRow struct {
    Page       string `json:"page"`
    Button     string `json:"button"`
    DurationMS int    `json:"duration_ms"`
}

page, err := wavehouse.FetchTyped[ClickRow](ctx,
    client.From("clicks").Select("page", "button", "duration_ms").Limit(100),
)
// page.Data is []ClickRow

Batch Insert (NDJSON)

// Array of maps — serialized to NDJSON automatically.
result, _ := client.From("clicks").Insert(ctx, []map[string]any{
    {"page": "/a", "button": "cta"},
    {"page": "/b", "button": "nav"},
})
// result.OK, result.Total, result.Succeeded, result.Failed

// Pre-formatted NDJSON string.
result, _ = client.From("clicks").InsertNDJSON(ctx,
    `{"page":"/a"}`+"\n"+`{"page":"/b"}`,
)

Streaming (SSE)

stream := client.From("clicks").Stream(&wavehouse.StreamOptions{
    Since: "2026-01-01T00:00:00Z",
})
defer stream.Close()

// Channel-based consumption.
for event := range stream.Events() {
    fmt.Println(event.Table, event.Data)
}

// Or callback-based.
unsub := stream.Subscribe(&wavehouse.StreamSubscriber{
    Next:   func(e wavehouse.StreamEvent) { fmt.Println(e.Data) },
    Status: func(s wavehouse.StreamStatus) { fmt.Println("status:", s) },
})
defer unsub()

Live Queries

lq := client.From("clicks").
    SelectAll().
    OrderBy("received_timestamp", "desc").
    Limit(100).
    LiveQuery(&wavehouse.StreamSubscriber{
        Initial: func(rows []map[string]any, err error) {
            // Historical backfill.
            fmt.Println("initial rows:", len(rows))
        },
        Next: func(e wavehouse.StreamEvent) {
            // Live events after backfill.
            fmt.Println("live:", e.Data)
        },
    }, nil)
defer lq.Close()

Named Pipes

// Execute a pipe.
rows, _ := wavehouse.Fetch[map[string]any](ctx,
    client.Pipe("top_pages", map[string]any{"limit": 10}),
)

// Admin: manage pipes.
client.Pipes.Set(ctx, "top_pages", wavehouse.PipeDef{
    SQL:          "SELECT page, count() as views FROM clicks GROUP BY page LIMIT {{limit}}",
    AllowedRoles: []string{"viewer", "admin"},
})
pipes, _ := client.Pipes.List(ctx)
client.Pipes.Delete(ctx, "old_pipe")

Admin

// Schema introspection (admin-only).
schemas, _ := client.Schema.List(ctx)
client.Schema.Refresh(ctx)

// Policy management (admin-only).
policy, _ := client.Policy.Get(ctx)
client.Policy.Set(ctx, policy)
result, _ := client.Policy.Validate(ctx, policy)

// DLQ stats (admin-only).
stats, _ := client.DLQ.List(ctx)

// Raw SQL (admin-only).
rows, _ := wavehouse.SQL[map[string]any](ctx, client, "SELECT count() FROM clicks")

Codegen

Generate Go structs from a running WaveHouse instance:

export WAVEHOUSE_AUTH=<admin-jwt>   # avoids leaking the token via argv
go run github.com/Wave-RF/WaveHouse/clients/go/cmd/wavehouse-codegen@latest \
    --url http://localhost:8080 \
    --out ./db_types.go \
    --package myapp

See the full type mapping in the docs.

Error Handling

Every request-response operation (queries, ingest, pipes, admin) returns (T, error) — or a bare error for operations with no result body (Pipes.Set/Delete, Policy.Set, Schema.Refresh, Sys.Health). Errors originating from the HTTP exchange are *wavehouse.Error — unwrap with errors.As. Client-side failures before a request goes out (an Auth provider error, a request-body marshal failure) are plain wrapped errors, so handle the errors.As == false case too. Streaming lifecycle methods (Stream, Subscribe, Close, and Connected) deliver errors through callbacks or plain errors instead:

page, err := client.From("clicks").Fetch(ctx)
if err != nil {
    var whErr *wavehouse.Error
    if errors.As(err, &whErr) {
        fmt.Println(whErr.Status, whErr.Code, whErr.Message, whErr.Retryable)
    }
}

The HTTP layer retries 5xx, 429, and network errors with exponential backoff (default 2 retries). Retry-After on a 503 or 429 is honored, capped at 30s. Context cancellation returns immediately with code ABORTED.

License

Apache-2.0

Documentation

Overview

Package wavehouse is the official Go SDK for WaveHouse — a schema-aware real-time API gateway for ClickHouse. Zero third-party runtime dependencies.

Create a client with NewClient, then use Client.From for table operations, Client.Pipe for named queries, or the admin namespaces (Client.Schema, Client.Policy, etc.) for management. 

client := wavehouse.NewClient(wavehouse.Config{
    BaseURL: "http://localhost:8080",
})
rows, err := client.From("clicks").SelectAll().FetchUntyped(ctx)

Index

Examples

Constants

View Source 
const DefaultLimit = 1000

DefaultLimit is applied when no explicit limit is set — deliberately tighter than the backend's DefaultMaxRows (10000) safety cap.

Variables

This section is empty.

Functions

func Fetch 

func Fetch[Row any](ctx context.Context, p *PipeRef) ([]Row, error)

Fetch executes the pipe and returns the result rows decoded into []T.

func IsRetryable 

func IsRetryable(err error) bool

IsRetryable reports whether err wraps a retryable *Error.

func SQL 

func SQL[Row any](ctx context.Context, c *Client, query string) ([]Row, error)

SQL executes a raw SQL query against ClickHouse. Requires the admin role. The server proxies the SQL verbatim to ClickHouse's HTTP interface. Results are decoded into []T; use [map[string]any] for dynamic schemas.

Example 
package main

import (
	"context"
	"fmt"
	"log"

	wavehouse "github.com/Wave-RF/WaveHouse/clients/go"
)

func main() {
	client := wavehouse.NewClient(wavehouse.Config{
		BaseURL: "http://localhost:8080",
		Auth:    wavehouse.StaticToken("admin-token"),
	})

	rows, err := wavehouse.SQL[map[string]any](
		context.Background(), client,
		"SELECT page, count() as views FROM clicks GROUP BY page LIMIT 5",
	)
	if err != nil {
		log.Fatal(err)
	}
	for _, row := range rows {
		fmt.Println(row["page"], row["views"])
	}
}

func StaticToken 

func StaticToken(token string) func(context.Context) (string, error)

StaticToken returns an Auth function that always returns the same token. Convenience for cases where the token doesn't rotate.

Types

type Aggregation 

type Aggregation struct {
	Fn     string `json:"fn"`
	Column string `json:"column"`
	Alias  string `json:"alias"`
}

Aggregation describes a single aggregation (e.g. count, sum).

type Client 

type Client struct {

	// Schema provides admin-only schema introspection.
	Schema *SchemaNamespace
	// Policy provides admin-only access-control policy management.
	Policy *PolicyNamespace
	// DLQ provides admin-only dead-letter-queue statistics.
	DLQ *DLQNamespace
	// Sys provides system health checks.
	Sys *SysNamespace
	// Pipes provides admin-only named-pipe management.
	Pipes *PipesNamespace
	// contains filtered or unexported fields
}

Client is the WaveHouse SDK entry point.

func NewClient 

func NewClient(cfg Config) *Client

NewClient creates a new WaveHouse client.

Example

ExampleNewClient demonstrates creating an unauthenticated client and performing a health check. The Output assertion is omitted because the example needs a running server. 

package main

import (
	"context"
	"log"

	wavehouse "github.com/Wave-RF/WaveHouse/clients/go"
)

func main() {
	client := wavehouse.NewClient(wavehouse.Config{
		BaseURL: "http://localhost:8080",
	})

	// Health check — returns nil when the server is reachable.
	if err := client.Sys.Health(context.Background()); err != nil {
		log.Fatal(err)
	}
}
Example (WithAuth) 
package main

import (
	wavehouse "github.com/Wave-RF/WaveHouse/clients/go"
)

func main() {
	_ = wavehouse.NewClient(wavehouse.Config{
		BaseURL: "http://localhost:8080",
		Auth:    wavehouse.StaticToken("my-jwt-token"),
	})
}

func (*Client) From 

func (c *Client) From(table string) *TableRef

From returns a reference to a table for queries, inserts, and streams.

Example 
package main

import (
	"context"
	"fmt"
	"log"

	wavehouse "github.com/Wave-RF/WaveHouse/clients/go"
)

func main() {
	client := wavehouse.NewClient(wavehouse.Config{
		BaseURL: "http://localhost:8080",
	})

	// Insert a row.
	_, _ = client.From("clicks").Insert(context.Background(), map[string]any{
		"page":   "/home",
		"button": "cta",
	})

	// Query with the builder.
	page, err := client.From("clicks").
		Select("page", "button").
		Where("page", wavehouse.OpEq, "/home").
		OrderBy("page", "asc").
		Limit(10).
		FetchUntyped(context.Background())
	if err != nil {
		log.Fatal(err)
	}

	for _, row := range page.Data {
		fmt.Println(row["page"])
	}
}

func (*Client) Pipe 

func (c *Client) Pipe(name string, params map[string]any) *PipeRef

Pipe returns a reference to a named query pipe. Pass params for the pipe's template parameters.

type ClientOptions 

type ClientOptions struct {
	// MaxRetries is the maximum number of retry attempts for retryable errors.
	// Total attempts = MaxRetries + 1. Default: 2.
	MaxRetries int
}

ClientOptions tunes transport behavior.

type Column 

type Column struct {
	Name       string `json:"name"`
	Type       string `json:"type"`
	IsNullable bool   `json:"is_nullable"`
	HasDefault bool   `json:"has_default"`
}

Column describes a single column in a table schema.

type Config 

type Config struct {
	// BaseURL of the WaveHouse server (e.g. "http://localhost:8080").
	BaseURL string

	// Auth provides a bearer token for authenticated requests. Called before
	// each request; return "" to skip the Authorization header. Nil means
	// unauthenticated access (the server falls back to default_role).
	Auth func(ctx context.Context) (string, error)

	// Options tunes transport behavior.
	Options *ClientOptions

	// HTTPClient overrides the default http.Client. Useful for custom TLS,
	// proxies, or test transports.
	HTTPClient *http.Client
}

Config configures a Client.

type DLQNamespace 

type DLQNamespace struct {
	// contains filtered or unexported fields
}

DLQNamespace provides admin-only dead-letter-queue statistics.

func (*DLQNamespace) List 

func (d *DLQNamespace) List(ctx context.Context) (*DLQStats, error)

List returns DLQ statistics (message counts per table). Admin-only.

func (*DLQNamespace) Stream 

func (d *DLQNamespace) Stream(opts *StreamOptions) *StreamController

Stream subscribes to live DLQ events. Not yet functional server-side (#197).

func (*DLQNamespace) Table 

func (d *DLQNamespace) Table(ctx context.Context, name string) (*DLQStats, error)

Table returns DLQ stats filtered by table name. Admin-only.

type DLQStats 

type DLQStats struct {
	Tables map[string]int `json:"tables"`
	Total  int            `json:"total"`
}

DLQStats describes dead-letter-queue statistics.

type Error 

type Error struct {
	// Status is the HTTP status code (0 for network/abort errors).
	Status int `json:"status"`
	// Code is a machine-readable error code (e.g. "HTTP_400", "NETWORK_ERROR", "ABORTED").
	Code string `json:"code"`
	// Message is a human-readable description.
	Message string `json:"message"`
	// Details contains the full parsed error body, if available.
	Details map[string]any `json:"details,omitempty"`
	// Retryable indicates whether the request can be retried.
	Retryable bool `json:"retryable"`
}

Error is the structured error returned by all SDK operations. Use errors.As to extract it from wrapped errors.

func (*Error) Error 

func (e *Error) Error() string

type FilterOp 

type FilterOp string

FilterOp is an SDK-facing filter operator. 

const (
	OpEq      FilterOp = "="
	OpNeq     FilterOp = "!="
	OpGt      FilterOp = ">"
	OpGte     FilterOp = ">="
	OpLt      FilterOp = "<"
	OpLte     FilterOp = "<="
	OpIn      FilterOp = "in"
	OpLike    FilterOp = "like"
	OpNotLike FilterOp = "not_like"
)

type InsertRecordResult 

type InsertRecordResult struct {
	Index     int    `json:"index"`
	OK        *bool  `json:"ok,omitempty"`
	Duplicate *bool  `json:"duplicate,omitempty"`
	Error     string `json:"error,omitempty"`
}

InsertRecordResult is a per-record outcome from a batch insert.

type InsertResult 

type InsertResult struct {
	OK         bool                 `json:"ok"`
	Duplicate  *bool                `json:"duplicate,omitempty"`
	Total      *int                 `json:"total,omitempty"`
	Succeeded  *int                 `json:"succeeded,omitempty"`
	Failed     *int                 `json:"failed,omitempty"`
	Duplicates *int                 `json:"duplicates,omitempty"`
	Results    []InsertRecordResult `json:"results,omitempty"`
}

InsertResult is the outcome of an insert operation.

type LiveQueryHandle 

type LiveQueryHandle struct {
	// contains filtered or unexported fields
}

LiveQueryHandle controls a live query that combines historical backfill with a real-time stream.

func (*LiveQueryHandle) Close 

func (lq *LiveQueryHandle) Close()

Close shuts down the live query and the underlying stream. The close state is applied synchronously: no new subscriber callbacks start after Close returns (a callback already in flight may still complete).

type OrderClause 

type OrderClause struct {
	Column string `json:"column"`
	Dir    string `json:"dir"` // "asc" or "desc"
}

OrderClause describes a single ORDER BY clause.

type Page 

type Page[T any] struct {
	// Data is the result rows.
	Data []T
	// HasMore is true if more rows may be available.
	HasMore bool
	// Next fetches the next page. Nil when no cursor is available.
	Next func(ctx context.Context) (*Page[T], error)
}

Page wraps a result set with pagination metadata.

func FetchTyped 

func FetchTyped[Row any](ctx context.Context, q *QueryBuilder) (*Page[Row], error)

FetchTyped executes the query and decodes rows into []T.

type ParamDef 

type ParamDef struct {
	Name     string `json:"name"`
	Type     string `json:"type"`
	Required bool   `json:"required,omitempty"`
	Default  any    `json:"default,omitempty"`
}

ParamDef describes a pipe parameter.

type Pipe 

type Pipe struct {
	Name         string     `json:"name"`
	SQL          string     `json:"sql"`
	Parameters   []ParamDef `json:"parameters,omitempty"`
	Description  string     `json:"description,omitempty"`
	AllowedRoles []string   `json:"allowed_roles,omitempty"`
}

Pipe describes a named query pipe definition.

type PipeDef 

type PipeDef struct {
	SQL          string     `json:"sql"`
	Parameters   []ParamDef `json:"parameters,omitempty"`
	Description  string     `json:"description,omitempty"`
	AllowedRoles []string   `json:"allowed_roles,omitempty"`
}

PipeDef is the definition body for creating/updating a pipe (Pipe minus name).

type PipeRef 

type PipeRef struct {
	// contains filtered or unexported fields
}

PipeRef is a reference to a named query pipe. Use Fetch to execute it.

func (*PipeRef) FetchUntyped 

func (p *PipeRef) FetchUntyped(ctx context.Context) ([]map[string]any, error)

FetchUntyped executes the pipe and returns rows as []map[string]any.

func (*PipeRef) Stream 

func (p *PipeRef) Stream(opts *StreamOptions) *StreamController

Stream opens a live event stream from the pipe's underlying query.

This streams by table name, using the pipe's own name as the table — it only works when the pipe name is also a valid table name. This matches the TS SDK's PipeRef.stream(), which has the same limitation.

type PipesNamespace 

type PipesNamespace struct {
	// contains filtered or unexported fields
}

PipesNamespace provides admin-only named-pipe management.

func (*PipesNamespace) Delete 

func (p *PipesNamespace) Delete(ctx context.Context, name string) error

Delete removes a pipe by name. Admin-only.

func (*PipesNamespace) Get 

func (p *PipesNamespace) Get(ctx context.Context, name string) (*Pipe, error)

Get returns a single pipe definition by name. Admin-only.

func (*PipesNamespace) List 

func (p *PipesNamespace) List(ctx context.Context) ([]Pipe, error)

List returns all registered pipes. Admin-only.

func (*PipesNamespace) Set 

func (p *PipesNamespace) Set(ctx context.Context, name string, def PipeDef) error

Set creates or updates a pipe. Admin-only.

type Policy 

type Policy struct {
	DefaultRole string `json:"default_role,omitempty"`
	// AdminRole is the role granted full access and the allowlist bypass.
	// Empty means the server's default ("admin") applies.
	AdminRole string                 `json:"admin_role,omitempty"`
	Tables    map[string]TablePolicy `json:"tables"`
}

Policy describes the server's access-control policy.

type PolicyFilter 

type PolicyFilter struct {
	Eq  *string `json:"_eq,omitempty"`
	Neq *string `json:"_neq,omitempty"`
	Gt  *string `json:"_gt,omitempty"`
	Lt  *string `json:"_lt,omitempty"`
	In  *string `json:"_in,omitempty"`
}

PolicyFilter describes a policy filter predicate. Fields are pointers with omitempty so an intentional empty-string comparison (e.g. Eq pointing at "") is sent as "", while an unset operator is omitted entirely — never null — matching the server's absent-operator semantics.

type PolicyNamespace 

type PolicyNamespace struct {
	// contains filtered or unexported fields
}

PolicyNamespace provides admin-only access-control policy management.

func (*PolicyNamespace) Get 

func (p *PolicyNamespace) Get(ctx context.Context) (*Policy, error)

Get returns the current access-control policy. Admin-only.

func (*PolicyNamespace) Set 

func (p *PolicyNamespace) Set(ctx context.Context, pol *Policy) error

Set replaces the entire access-control policy. Admin-only.

func (*PolicyNamespace) Validate 

func (p *PolicyNamespace) Validate(ctx context.Context, pol *Policy) (*ValidationResult, error)

Validate checks a policy without applying it (dry run). Admin-only.

type QueryBuilder 

type QueryBuilder struct {
	// contains filtered or unexported fields
}

QueryBuilder builds structured queries. Immutable — every chain method returns a new builder. Use Fetch or FetchUntyped to execute.

func (*QueryBuilder) Aggregate 

func (q *QueryBuilder) Aggregate(fn, column, alias string) *QueryBuilder

Aggregate adds a custom aggregation function.

func (*QueryBuilder) Avg 

func (q *QueryBuilder) Avg(column, alias string) *QueryBuilder

Avg adds an AVG aggregation.

func (*QueryBuilder) CacheTTL 

func (q *QueryBuilder) CacheTTL(seconds int) *QueryBuilder

CacheTTL records a desired result-cache TTL. Currently client-side only — the server derives TTLs adaptively (#280).

func (*QueryBuilder) Count 

func (q *QueryBuilder) Count(column, alias string) *QueryBuilder

Count adds a COUNT aggregation.

func (*QueryBuilder) CountDistinct 

func (q *QueryBuilder) CountDistinct(column, alias string) *QueryBuilder

CountDistinct adds a COUNT DISTINCT aggregation.

func (*QueryBuilder) FetchUntyped 

func (q *QueryBuilder) FetchUntyped(ctx context.Context) (*Page[map[string]any], error)

FetchUntyped executes the query and returns rows as []map[string]any.

func (*QueryBuilder) GroupBy 

func (q *QueryBuilder) GroupBy(columns ...string) *QueryBuilder

GroupBy appends columns to the GROUP BY clause.

func (*QueryBuilder) Limit 

func (q *QueryBuilder) Limit(n int) *QueryBuilder

Limit sets the maximum number of rows to return.

func (*QueryBuilder) LiveQuery 

func (q *QueryBuilder) LiveQuery(sub *StreamSubscriber, opts *StreamOptions) *LiveQueryHandle

LiveQuery starts a live query: fetches historical data, then streams live updates. The subscriber's Initial is called once, then Next fires for each live event. Returns a LiveQuery handle with a Close method.

func (*QueryBuilder) Max 

func (q *QueryBuilder) Max(column, alias string) *QueryBuilder

Max adds a MAX aggregation.

func (*QueryBuilder) Min 

func (q *QueryBuilder) Min(column, alias string) *QueryBuilder

Min adds a MIN aggregation.

func (*QueryBuilder) OrderBy 

func (q *QueryBuilder) OrderBy(column, dir string) *QueryBuilder

OrderBy appends an ORDER BY clause. dir defaults to "asc".

func (*QueryBuilder) Select 

func (q *QueryBuilder) Select(columns ...string) *QueryBuilder

Select appends columns to the projection.

func (*QueryBuilder) SelectAll 

func (q *QueryBuilder) SelectAll() *QueryBuilder

SelectAll requests every column the caller's role may read.

func (*QueryBuilder) Stream 

func (q *QueryBuilder) Stream(opts *StreamOptions) *StreamController

Stream opens a live SSE event stream for this query's table. Filters and column projections are applied client-side.

func (*QueryBuilder) Sum 

func (q *QueryBuilder) Sum(column, alias string) *QueryBuilder

Sum adds a SUM aggregation.

func (*QueryBuilder) TimeRange 

func (q *QueryBuilder) TimeRange(column, since, until string) *QueryBuilder

TimeRange filters by a time window. since and until accept RFC3339 timestamps or relative durations ("1h", "30m", "7d", "2w").

func (*QueryBuilder) Where 

func (q *QueryBuilder) Where(column string, op FilterOp, value any) *QueryBuilder

Where adds a filter condition.

type QueryFilter 

type QueryFilter struct {
	Column string `json:"column"`
	Op     string `json:"op"`
	Value  any    `json:"value"`
}

QueryFilter describes a single WHERE condition.

type RolePermissions 

type RolePermissions struct {
	AllowColumns        []string                `json:"allow_columns,omitempty"`
	DenyColumns         []string                `json:"deny_columns,omitempty"`
	Filter              map[string]PolicyFilter `json:"filter,omitempty"`
	Check               map[string]PolicyFilter `json:"check,omitempty"`
	AllowedAggregations []string                `json:"allowed_aggregations,omitempty"`
	DeniedAggregations  []string                `json:"denied_aggregations,omitempty"`
	MaxRows             *int                    `json:"max_rows,omitempty"`
	MaxExecutionTime    any                     `json:"max_execution_time,omitempty"`
	MaxRowsToRead       *int64                  `json:"max_rows_to_read,omitempty"`
	MaxMemoryUsage      any                     `json:"max_memory_usage,omitempty"`
}

RolePermissions describes a role's access to a table.

type SchemaNamespace 

type SchemaNamespace struct {
	// contains filtered or unexported fields
}

SchemaNamespace provides admin-only schema introspection.

func (*SchemaNamespace) List 

func (s *SchemaNamespace) List(ctx context.Context) (Schemas, error)

List returns all table schemas discovered from ClickHouse. Admin-only.

func (*SchemaNamespace) Refresh 

func (s *SchemaNamespace) Refresh(ctx context.Context) error

Refresh forces a schema re-discovery from ClickHouse. Admin-only.

type Schemas 

type Schemas map[string]TableSchema

Schemas maps table names to their schemas.

type StreamController 

type StreamController struct {
	// contains filtered or unexported fields
}

StreamController manages a live SSE event stream. Use Subscribe for callback-based consumption or Events for channel-based consumption.

func (*StreamController) Close 

func (sc *StreamController) Close()

Close shuts down the stream and releases resources. Non-blocking so it is safe to call from subscriber callbacks (which run on the stream goroutine).

func (*StreamController) Connected 

func (sc *StreamController) Connected(ctx context.Context) error

Connected blocks until the stream reaches "live" status or the context expires. Returns an error if the stream closes before connecting.

func (*StreamController) Events 

func (sc *StreamController) Events() <-chan StreamEvent

Events returns a read-only channel that receives stream events. The channel is closed when the stream closes. Events buffer into it from stream construction (matching the TS SDK), so events that arrive before the first Events() call are not lost. A Subscribe-only consumer that never calls Events() at most fills the 256-slot buffer and trips the one-time drop log.

func (*StreamController) Status 

func (sc *StreamController) Status() StreamStatus

Status returns the current connection status.

func (*StreamController) Subscribe 

func (sc *StreamController) Subscribe(sub *StreamSubscriber) func()

Subscribe registers callbacks for stream events. Returns an unsubscribe function. The subscriber's Status callback fires immediately with the current status.

type StreamEvent 

type StreamEvent struct {
	Table     string         `json:"table"`
	Timestamp string         `json:"timestamp"`
	Data      map[string]any `json:"data"`
}

StreamEvent is a single event from an SSE stream.

type StreamOptions 

type StreamOptions struct {
	// Since is an RFC3339 timestamp for gap-fill replay.
	Since string
}

StreamOptions configures a stream.

type StreamStatus 

type StreamStatus string

StreamStatus represents the connection state of a stream. 

const (
	StatusConnecting   StreamStatus = "connecting"
	StatusLive         StreamStatus = "live"
	StatusReconnecting StreamStatus = "reconnecting"
	StatusClosed       StreamStatus = "closed"
)

type StreamSubscriber 

type StreamSubscriber struct {
	// Initial is called once with historical backfill data (live queries only).
	Initial func(rows []map[string]any, err error)
	// Next is called for each live event.
	Next func(event StreamEvent)
	// Status is called when the connection status changes.
	Status func(status StreamStatus)
	// Error is called on stream errors.
	Error func(err error)
}

StreamSubscriber receives events from a stream.

type StructuredQuery 

type StructuredQuery struct {
	// Columns to project. A literal "*" is a column named "*", not a wildcard.
	// Omitting columns (with no aggregations and no select_all) selects nothing.
	Columns []string `json:"columns,omitempty"`
	// SelectAll requests every column the caller's role may read.
	// Mutually exclusive with a non-empty Columns list.
	SelectAll bool `json:"select_all,omitempty"`
	// Aggregations (count, sum, avg, etc.).
	Aggregations []Aggregation `json:"aggregations,omitempty"`
	// Filters (WHERE conditions, ANDed).
	Filters []QueryFilter `json:"filters,omitempty"`
	// GroupBy columns.
	GroupBy []string `json:"group_by,omitempty"`
	// OrderBy clauses.
	OrderBy []OrderClause `json:"order_by,omitempty"`
	// Limit caps the result set.
	Limit *int `json:"limit,omitempty"`
	// TimeRange filters by a time window.
	TimeRange *TimeRange `json:"time_range,omitempty"`
}

StructuredQuery is the wire format for POST /v1/query.

type SysNamespace 

type SysNamespace struct {
	// contains filtered or unexported fields
}

SysNamespace provides system health checks.

func (*SysNamespace) Health 

func (s *SysNamespace) Health(ctx context.Context) error

Health pings the server's public /v1/health endpoint. Returns nil when the server is reachable and past boot, or an error describing the failure.

type TablePolicy 

type TablePolicy struct {
	Select map[string]RolePermissions `json:"select,omitempty"`
	Insert map[string]RolePermissions `json:"insert,omitempty"`
}

TablePolicy describes per-table access control.

type TableRef 

type TableRef struct {
	// contains filtered or unexported fields
}

TableRef is a reference to a table. Use it for queries, inserts, schema, and streams. Safe for concurrent use: it holds no mutable state, and every builder method returns a fresh value.

func (*TableRef) Fetch 

func (t *TableRef) Fetch(ctx context.Context) (*Page[map[string]any], error)

Fetch is a SELECT * shortcut with a default limit of 1000.

func (*TableRef) Insert 

func (t *TableRef) Insert(ctx context.Context, data any) (*InsertResult, error)

Insert inserts one or more rows into this table. A single map or struct is sent as JSON; any slice — []map[string]any, a generated/user-defined row type such as []ClickRow, etc. — is serialized to NDJSON for batch ingest.

func (*TableRef) InsertNDJSON 

func (t *TableRef) InsertNDJSON(ctx context.Context, ndjson string) (*InsertResult, error)

InsertNDJSON inserts pre-formatted NDJSON (one record per line).

func (*TableRef) Schema 

func (t *TableRef) Schema(ctx context.Context) (*TableSchema, error)

Schema returns the table's column definitions from ClickHouse. Admin-only.

func (*TableRef) Select 

func (t *TableRef) Select(columns ...string) *QueryBuilder

Select starts building a typed query with the given column projection.

func (*TableRef) SelectAll 

func (t *TableRef) SelectAll() *QueryBuilder

SelectAll starts a query that selects every column the caller's role may read.

func (*TableRef) Stream 

func (t *TableRef) Stream(opts *StreamOptions) *StreamController

Stream opens a live SSE event stream for this table.

type TableSchema 

type TableSchema struct {
	Name    string   `json:"name"`
	Columns []Column `json:"columns"`
}

TableSchema describes a table's schema.

type TimeRange 

type TimeRange struct {
	Column string `json:"column"`
	Since  string `json:"since"`
	Until  string `json:"until,omitempty"`
}

TimeRange filters by a time window on a column.

type ValidationResult 

type ValidationResult struct {
	Valid bool `json:"valid"`
}

ValidationResult is the response from policy validation.

Source Files

View all Source files

Directories

Path Synopsis
cmd
wavehouse-codegen command
Command wavehouse-codegen reads a WaveHouse server's /v1/schema endpoint and generates Go struct definitions for use with the wavehouse SDK.
 Command wavehouse-codegen reads a WaveHouse server's /v1/schema endpoint and generates Go struct definitions for use with the wavehouse SDK.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.