api

package
v0.1.0 Latest Latest
Warning

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

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

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func EnsureDLQStream 

func EnsureDLQStream(ctx context.Context, js jetstream.JetStream, maxBytes int64) error

EnsureDLQStream creates the DLQ JetStream stream if it doesn't exist.

func NewRouter 

func NewRouter(deps Dependencies) http.Handler

NewRouter creates the chi router with all routes.

func RequireAdmin 

func RequireAdmin(store *policy.Store, logger *slog.Logger) func(http.Handler) http.Handler

RequireAdmin restricts a route to the policy admin role (policy.AdminRole — configurable via admin_role, "admin" by default). The role established by the auth middleware and the live policy are read per request, so an admin_role change applies without a restart. A nil policy (none configured yet, or deleted from KV) admits nobody via a role — IsAdmin(nil) is false — so a role-based caller can't bootstrap by writing the first policy over this gate. The exception is the operator key: auth.IsOperator passes this gate even under a nil policy, so an operator can restore a wiped policy over HTTP (break-glass).

Authentication is decoupled from this gate: a missing/invalid/expired token resolves to an empty (non-admin) role and is denied here. Denials go through writeAuthzDenied, so a present-but-invalid token fails loud (401 + token reason) rather than as a bare 403.

Types

type BootState 

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

BootState tracks a one-shot startup diagnostic surfaced by /livez. While Err() returns non-nil the binary is considered to be in degraded-boot mode: /livez responds 503 with the diagnostic message instead of 200, so an operator can curl the endpoint to learn why the gateway isn't accepting traffic yet. Once boot work (today: ClickHouse schema discovery) succeeds, Set(nil) flips /livez back to 200.

BootState is safe for concurrent use.

func NewBootState 

func NewBootState(initialErr error) *BootState

NewBootState returns a BootState seeded with initialErr. Pass nil if the binary is fully ready at construction time; pass a non-nil error to start in degraded mode (the goroutine that performs boot work calls Set(nil) on success).

func (*BootState) Err 

func (b *BootState) Err() error

Err returns the current diagnostic, or nil if the binary is ready.

func (*BootState) Set 

func (b *BootState) Set(err error)

Set replaces the current diagnostic. Pass nil to mark the binary ready.

type DLQHandler 

type DLQHandler struct {
	JS     jetstream.JetStream
	Logger *slog.Logger
}

DLQHandler exposes Dead Letter Queue statistics.

func NewDLQHandler 

func NewDLQHandler(js jetstream.JetStream, logger *slog.Logger) *DLQHandler

func (*DLQHandler) Stats 

func (h *DLQHandler) Stats(w http.ResponseWriter, r *http.Request)

Stats returns per-table message counts in the DLQ stream. Supports optional ?table= query parameter to filter by table name.

type Dependencies 

type Dependencies struct {
	Ingest          *IngestHandler
	Query           *QueryHandler
	SSE             *StreamHandler
	Health          *HealthHandler
	Version         *VersionHandler
	Schema          *SchemaHandler
	DLQ             *DLQHandler
	Policy          *PolicyHandler
	Pipes           *PipesHandler
	StructuredQuery *StructuredQueryHandler
	AuthMW          func(http.Handler) http.Handler
	// PolicyStore backs the RequireAdmin gate: the admin role (policy.AdminRole)
	// is read live from the policy, so admin_role changes apply without a restart.
	PolicyStore *policy.Store
	JS          jetstream.JetStream // for SSE gap-fill
	CORSOrigins []string            // allowed CORS origins; ["*"] = allow all
	Logger      *slog.Logger
	// MetricsHandler, if non-nil, is mounted at MetricsPath as an unauthenticated
	// endpoint (Prometheus convention). Wired by main.go from the OTel Prometheus
	// exporter when observability.metrics.prometheus.enabled is true AND port is 0.
	MetricsHandler http.Handler
	MetricsPath    string
}

Dependencies holds all handler dependencies.

type HealthHandler 

type HealthHandler struct {
	CHConn driver.Conn
	// Boot is consulted by both Liveness and Readiness. When non-nil and
	// its Err() is non-nil, both endpoints report 503 with the diagnostic
	// — used while boot-time schema discovery is still failing in the
	// retry loop. A nil Boot preserves the pre-retry-loop behaviour
	// (Liveness always 200; Readiness 503 only on Ping failure).
	Boot *BootState
}

HealthHandler provides liveness and readiness probes.

func NewHealthHandler 

func NewHealthHandler(chConn driver.Conn) *HealthHandler

func (*HealthHandler) Liveness 

func (h *HealthHandler) Liveness(w http.ResponseWriter, _ *http.Request)

func (*HealthHandler) Online 

func (h *HealthHandler) Online(w http.ResponseWriter, _ *http.Request)

Online is a content-free public liveness ping served at /v1/health for the SDK's "is this server reachable / accepting data" check (and for picking among servers in a distributed setup). It mirrors Liveness's status logic — 200 once boot completes, 503 while boot-time schema discovery is still failing — but writes no body: the caller only branches on the status code, so there's nothing to JSON-encode or cache per request.

It deliberately lives under /v1 rather than reusing /livez: /livez (and /readyz, /healthz) are Kubernetes probe paths an operator may filter out at the reverse proxy, whereas /v1/health is documented public API surface the SDK can rely on staying reachable. It does NOT ping ClickHouse — readiness- based load balancing is the proxy/LB's job (via /readyz), not the client's.

func (*HealthHandler) Readiness 

func (h *HealthHandler) Readiness(w http.ResponseWriter, r *http.Request)

type IngestHandler 

type IngestHandler struct {
	Registry    *discovery.SchemaRegistry
	Dedup       dedupe.Deduplicator // nil if dedup disabled
	IDField     string              // dedup key field name (e.g. "event_id")
	RequireID   bool                // reject rows missing IDField instead of publishing un-deduped (dedupe.require_id)
	Publisher   mq.Publisher
	PolicyStore *policy.Store
	// contains filtered or unexported fields
}

IngestHandler handles POST /v1/ingest?table={table}

func NewIngestHandler 

func NewIngestHandler(registry *discovery.SchemaRegistry, pub mq.Publisher, logger *slog.Logger) *IngestHandler

func (*IngestHandler) Handle 

func (h *IngestHandler) Handle(w http.ResponseWriter, r *http.Request)

type PipesHandler 

type PipesHandler struct {
	Store       *pipes.Store
	PolicyStore *policy.Store // resolves empty role to default_role; may be nil
	CHConn      driver.Conn
	Cache       cache.Cache
	// contains filtered or unexported fields
}

PipesHandler handles named query pipe endpoints.

func NewPipesHandler 

func NewPipesHandler(store *pipes.Store, policyStore *policy.Store, conn driver.Conn, c cache.Cache, queryTimeout time.Duration, logger *slog.Logger) *PipesHandler

func (*PipesHandler) Delete 

func (h *PipesHandler) Delete(w http.ResponseWriter, r *http.Request)

Delete removes a named query (admin endpoint).

func (*PipesHandler) Execute 

func (h *PipesHandler) Execute(w http.ResponseWriter, r *http.Request)

Execute runs a named query with the provided parameters.

func (*PipesHandler) Get 

func (h *PipesHandler) Get(w http.ResponseWriter, r *http.Request)

Get returns a specific named query (admin endpoint).

func (*PipesHandler) List 

func (h *PipesHandler) List(w http.ResponseWriter, _ *http.Request)

List returns all named queries (admin endpoint).

func (*PipesHandler) Put 

func (h *PipesHandler) Put(w http.ResponseWriter, r *http.Request)

Put creates or updates a named query (admin endpoint).

type PolicyHandler 

type PolicyHandler struct {
	Store *policy.Store
	// contains filtered or unexported fields
}

PolicyHandler handles policy CRUD endpoints.

func NewPolicyHandler 

func NewPolicyHandler(store *policy.Store) *PolicyHandler

func (*PolicyHandler) Get 

func (h *PolicyHandler) Get(w http.ResponseWriter, r *http.Request)

Get returns the current access control policy.

func (*PolicyHandler) Put 

func (h *PolicyHandler) Put(w http.ResponseWriter, r *http.Request)

Put replaces the current access control policy.

func (*PolicyHandler) Validate 

func (h *PolicyHandler) Validate(w http.ResponseWriter, r *http.Request)

Validate checks a policy without saving it.

type QueryHandler 

type QueryHandler struct {
	HTTPClient *http.Client
	// Endpoint is the ClickHouse HTTP base URL (e.g.
	// `http://localhost:8123`). The handler appends query-string params
	// (`default_format`, `database`, `date_time_output_format`) per request
	// and POSTs the SQL as the request body.
	Endpoint string
	Username string
	Password string
	Database string
	// contains filtered or unexported fields
}

QueryHandler handles POST /v1/ops/query.

Authorization is enforced at the router (the /v1/ops/* RequireAdmin gate in NewRouter). The handler trusts any caller that reaches it. See internal/api/router.go for the role-gate rationale.

Implementation: a thin proxy to ClickHouse's HTTP interface. The SQL string is forwarded verbatim with `default_format=JSON`, ClickHouse returns either a `{"meta":..., "data":[...], ...}` JSON object for read queries or an empty body for mutations, and the handler emits just the `data` array (or `[]` for mutations) back to the caller.

Why a proxy instead of clickhouse-go's native Query/Exec:

  • ClickHouse classifies statements natively, so any single statement (arbitrary DDL/DML verbs, current and future) and inline FORMAT directives all just work without WaveHouse-side parsing. Multi-statement input (`SELECT 1; TRUNCATE t`) also works when the upstream ClickHouse has multi-query enabled, which is the default in recent versions; older or restrictively-configured servers may reject the second statement with a clear error.
  • There is no isMutation heuristic to maintain — no leading-verb table, no comment stripper, no CTE-aware paren scanner, no class of bug where a future ClickHouse verb routes the wrong way.
  • ClickHouse's own error messages reach the admin verbatim, which is exactly what they want from an escape hatch.
  • The cache (TieredCache + singleflight) was already removed in an earlier commit; this completes the simplification.

func NewQueryHandler 

func NewQueryHandler(endpoint, username, password, database string, queryTimeout time.Duration) *QueryHandler

NewQueryHandler builds a handler that proxies to ClickHouse over HTTP. endpoint should be the base URL (`http://host:8123`); username/password are forwarded via ClickHouse's `X-ClickHouse-User` / `X-ClickHouse-Key` headers (matching the ingest worker's convention in internal/ingest). database is set as the `?database=` query-string parameter when non-empty.

The HTTP client itself has no Timeout — every request gets a queryTimeout deadline from a context derived from the inbound request (see Handle), which bounds the whole exchange including body read. Setting `Timeout` here too would just duplicate that bound (and silently truncate any inbound context longer than queryTimeout).

func (*QueryHandler) Handle 

func (h *QueryHandler) Handle(w http.ResponseWriter, r *http.Request)

type SchemaHandler 

type SchemaHandler struct {
	Registry *discovery.SchemaRegistry
}

SchemaHandler exposes the discovered ClickHouse table schemas.

func NewSchemaHandler 

func NewSchemaHandler(registry *discovery.SchemaRegistry) *SchemaHandler

func (*SchemaHandler) Get 

func (h *SchemaHandler) Get(w http.ResponseWriter, r *http.Request)

Get returns the schema for a single table.

func (*SchemaHandler) List 

func (h *SchemaHandler) List(w http.ResponseWriter, _ *http.Request)

List returns all discovered table schemas.

func (*SchemaHandler) Refresh 

func (h *SchemaHandler) Refresh(w http.ResponseWriter, r *http.Request)

Refresh forces an immediate schema refresh from ClickHouse.

type StreamHandler 

type StreamHandler struct {
	Hub         *stream.Hub
	JS          jetstream.JetStream
	Heartbeater *stream.Heartbeater
	Metrics     *stream.Metrics
}

StreamHandler handles GET /v1/stream

func NewStreamHandler 

func NewStreamHandler(hub *stream.Hub, js jetstream.JetStream) *StreamHandler

func (*StreamHandler) Handle 

func (h *StreamHandler) Handle(w http.ResponseWriter, r *http.Request)

type StructuredQueryHandler 

type StructuredQueryHandler struct {
	CHConn      driver.Conn
	Cache       cache.Cache
	Registry    *discovery.SchemaRegistry
	PolicyStore *policy.Store
	BucketSecs  int
	// contains filtered or unexported fields
}

StructuredQueryHandler handles POST /v1/query?table={table}

func NewStructuredQueryHandler 

func NewStructuredQueryHandler(
	conn driver.Conn,
	c cache.Cache,
	registry *discovery.SchemaRegistry,
	policyStore *policy.Store,
	bucketSecs int,
	queryTimeout time.Duration,
	defaultMaxRows int,
	logger *slog.Logger,
) *StructuredQueryHandler

func (*StructuredQueryHandler) Handle 

func (h *StructuredQueryHandler) Handle(w http.ResponseWriter, r *http.Request)

type VersionHandler 

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

VersionHandler serves the binary's build metadata at /version. The payload is fully static for the process lifetime (the ldflags-injected build info plus the Go runtime version), so it's marshaled once in NewVersionHandler and the cached bytes are written on each request — no per-request encoding.

func NewVersionHandler 

func NewVersionHandler(version, gitCommit, buildTime string) *VersionHandler

NewVersionHandler pre-marshals the build-info JSON from the ldflags-injected values (version, git commit, build time — see the Makefile / .goreleaser.yaml) plus runtime.Version(). Marshaling a struct of strings cannot fail, so the error is discarded.

func (*VersionHandler) Handle 

func (h *VersionHandler) Handle(w http.ResponseWriter, _ *http.Request)

Handle writes the pre-marshaled build metadata. No auth gate (wired among the public routes in NewRouter) — the values are non-sensitive and already in the startup logs.

Source Files

View all Source files

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.