feat: caching for local table/scope invalidation

clients/ts/src/query-builder.test.ts

@@ -183,15 +183,6 @@ describe('QueryBuilder', () => {
     expect(body.time_range).toEqual({ column: 'ts', since: '2026-01-01', until: '2026-02-01' });
   });
 
-  // --- CacheTTL ---
-
-  it('builds cacheTTL', async () => {
-    await builder().cacheTTL(300).fetch();
-
-    const body = JSON.parse(fetchSpy.mock.calls[0][1].body);
-    expect(body.cache_ttl).toBe(300);
-  });
-
   // --- Empty AST ---
 
   it('omits empty arrays and undefined fields from AST', async () => {
@@ -204,7 +195,6 @@ describe('QueryBuilder', () => {
     expect(body.group_by).toBeUndefined();
     expect(body.order_by).toEqual([{ column: 'received_timestamp', dir: 'desc' }]);
     expect(body.time_range).toBeUndefined();
-    expect(body.cache_ttl).toBeUndefined();
   });
 
   // --- Fetch endpoint ---
@@ -330,7 +320,6 @@ describe('QueryBuilder', () => {
       order_by: [{ column: 'total', dir: 'desc' }],
       limit: 50,
       time_range: { column: 'received_timestamp', since: '1h' },
-      cache_ttl: 60,
     });
   });
 });

clients/ts/src/query-builder.ts

@@ -213,7 +213,6 @@ export class QueryBuilder<Row = Record<string, unknown>> implements PromiseLike<
     }
     if (effectiveLimit != null) ast.limit = effectiveLimit;
     if (this._state.timeRange) ast.time_range = this._state.timeRange;
-    if (this._state.cacheTTL != null) ast.cache_ttl = this._state.cacheTTL;
     return ast;
   }
 

clients/ts/src/types.ts

@@ -77,7 +77,6 @@ export interface StructuredQuery {
   order_by?: OrderClause[];
   limit?: number;
   time_range?: TimeRange;
-  cache_ttl?: number;
 }
 
 export interface Aggregation {

cmd/wavehouse/main.go

@@ -249,8 +249,9 @@ func run() int {
 		logger.Error("cache init", "error", err)
 		return 1
 	}
-	tiered := cache.NewTiered(l1, nil)
-	defer func() { _ = tiered.Close() }()
+	// TODO: eventually this is where we can switch between ristretto, redis, tiered (both), etc
+	cache := l1
+	defer func() { _ = cache.Close() }()
 
 	// Policy store (NATS KV + optional file bootstrap).
 	policyStore, err := policy.NewStore(ctx, embeddedMQ.JetStream(), cfg.Policy.FilePath, logger)
@@ -281,6 +282,7 @@ func run() int {
 	ingestStream, err := ingest.StartIngestWorker(
 		ctx,
 		embeddedMQ.NatsConn(),
+		cache,
 		cfg.ClickHouse.Addr,
 		cfg.ClickHouse.HTTPPort, // Uses 8123 by default
 		cfg.ClickHouse.HTTPScheme,
@@ -345,7 +347,7 @@ func run() int {
 		queryScheme = "http"
 	}
 	queryEndpoint := fmt.Sprintf("%s://%s", queryScheme, net.JoinHostPort(queryHost, cfg.ClickHouse.HTTPPort))
-	queryHandler := api.NewQueryHandler(queryEndpoint, cfg.ClickHouse.Username, cfg.ClickHouse.Password, cfg.ClickHouse.Database)
+	queryHandler := api.NewQueryHandler(queryEndpoint, cfg.ClickHouse.Username, cfg.ClickHouse.Password, cfg.ClickHouse.Database, cfg.ClickHouse.QueryTimeout)
 
 	healthHandler := api.NewHealthHandler(chConn)
 	healthHandler.Boot = bootState
@@ -365,8 +367,8 @@ func run() int {
 		Schema:          api.NewSchemaHandler(registry),
 		DLQ:             dlqHandler,
 		Policy:          api.NewPolicyHandler(policyStore),
-		Pipes:           api.NewPipesHandler(pipesStore, chConn, tiered, time.Duration(cfg.Cache.DefaultTTL)*time.Second),
-		StructuredQuery: api.NewStructuredQueryHandler(chConn, tiered, time.Duration(cfg.Cache.DefaultTTL)*time.Second, registry, policyStore, cfg.Cache.TimestampBucketSeconds),
+		Pipes:           api.NewPipesHandler(pipesStore, chConn, cache, cfg.ClickHouse.QueryTimeout),
+		StructuredQuery: api.NewStructuredQueryHandler(chConn, cache, registry, policyStore, cfg.Cache.TimestampBucketSeconds, cfg.ClickHouse.QueryTimeout),
 		AuthMW: api.JWTAuthMiddleware(api.AuthConfig{
 			Enabled:   cfg.Auth.Enabled,
 			JWTSecret: cfg.Auth.JWTSecret,

config.yaml

@@ -43,6 +43,7 @@ clickhouse:
   database: default
   username: default
   password: ""
+  query_timeout: 30s
 
 mq:
   gap_window_minutes: 15
@@ -54,7 +55,6 @@ dedupe:
 
 cache:
   l1_max_cost: 67108864
-  default_ttl: 300
   timestamp_bucket_seconds: 60
 
 auth:

docs/src/content/docs/api.md

@@ -265,8 +265,7 @@ Executes a type-safe structured query against a table. The query AST is validate
     "column": "received_timestamp",
     "since": "1h",
     "until": ""
-  },
-  "cache_ttl": 60
+  }
 }
 ```
 
@@ -279,7 +278,6 @@ Executes a type-safe structured query against a table. The query AST is validate
 | `order_by` | object[] | No | ORDER BY clauses (`column`, `dir`). |
 | `limit` | int | No | Max rows. |
 | `time_range` | object | No | Time window (`column`, `since`, `until`). `since` can be relative ("1h", "30m") or RFC3339. |
-| `cache_ttl` | int | No | Override default cache TTL (seconds). |
 
 **Response:**
 
@@ -576,6 +574,7 @@ The message format used on NATS JetStream between ingest and the batch consumer:
 ```json
 {
   "table_name": "clicks",
+  "scope": "scope",
   "received_timestamp": "2026-03-24T12:00:00.123456789Z",
   "data": {
     "page": "/home",
@@ -588,6 +587,7 @@ The message format used on NATS JetStream between ingest and the batch consumer:
 | Field | Type | Description |
 | ----- | ---- | ----------- |
 | `table_name` | string | Target ClickHouse table (from URL). |
+| `scope` | string | Optional scope namespace used for subject routing/cache invalidation context. |
 | `received_timestamp` | string | RFC 3339 nano timestamp when WaveHouse received the event. |
 | `data` | object | The original flat JSON body. |
 

docs/src/content/docs/architecture.md

@@ -105,8 +105,8 @@ The API layer uses [Chi](https://github.com/go-chi/chi) for routing with standar
 
 ### `ingest/` — Bento Pipeline, DLQ & Sweeping
 
-- **bento.go** — `StartIngestWorker` launches a Bento-based ingest pipeline: a JetStream input (`jsInput`) reads from the `WAVEHOUSE` stream via a durable `buffer-consumer` pull consumer, batches events per table, and performs bulk INSERTs to ClickHouse. The pipeline is **insert-only**. The wire format `EventMessage` carries `{table_name, received_timestamp, data}` and nothing else; the worker validates the table name (via `safeIdentifierRe`) and the payload's presence, then bulk-INSERTs. The embedded NATS server runs with `DontListen: true` (`internal/mq/embedded.go`), so the only Publishers reachable on the `ingest.>` subjects are in-process Go code — today, only the HTTP `/v1/ingest/{table}` handler. Non-insert mutations (`DELETE`/`UPDATE`/`TRUNCATE`/…) must go through `POST /v1/admin/query` under the admin/service role — see the Query Path section below; when authentication is enabled, the `/v1/admin/*` middleware enforces that role check at the API layer, so non-admin callers never reach the proxy. With `auth.enabled=false` (or `auth.dev_mode=true`) the endpoint is intentionally open — the dev/test posture. Failed batches are routed to a DLQ output (`dlqOutput`) which publishes the inner data payload (`{"id":"abc","field":...}`) to `dlq.{table}` NATS subjects when DLQ is enabled.
-- **types.go** — `EventMessage` struct (TableName, ReceivedTimestamp, Data) and `BufferConsumerName` constant, shared across API handlers and the ingest pipeline.
+- **bento.go** — `StartIngestWorker` launches a Bento-based ingest pipeline: a JetStream input (`jsInput`) reads from the `WAVEHOUSE` stream via a durable `buffer-consumer` pull consumer, batches events per table, and performs bulk INSERTs to ClickHouse. The pipeline is **insert-only**. The wire format `EventMessage` carries `{table_name, scope, received_timestamp, data}` and nothing else; the worker validates the table name (via `safeIdentifierRe`) and the payload's presence, then bulk-INSERTs. The embedded NATS server runs with `DontListen: true` (`internal/mq/embedded.go`), so the only Publishers reachable on the `ingest.>` subjects are in-process Go code — today, only the HTTP `/v1/ingest/{table}` handler. Non-insert mutations (`DELETE`/`UPDATE`/`TRUNCATE`/…) must go through `POST /v1/admin/query` under the admin/service role — see the Query Path section below; when authentication is enabled, the `/v1/admin/*` middleware enforces that role check at the API layer, so non-admin callers never reach the proxy. With `auth.enabled=false` (or `auth.dev_mode=true`) the endpoint is intentionally open — the dev/test posture. Failed batches are routed to a DLQ output (`dlqOutput`) which publishes the inner data payload (`{"id":"abc","field":...}`) to `dlq.{table}` NATS subjects when DLQ is enabled.
+- **types.go** — `EventMessage` struct (TableName, Scope, ReceivedTimestamp, Data) and `BufferConsumerName` constant, shared across API handlers and the ingest pipeline.
 - **sweeper.go** — `Sweeper` implements the Active Sweeper pattern. It runs every minute and purges NATS JetStream messages that are **both** ACKed by the buffer consumer (written to ClickHouse) **and** older than the configurable gap window.
 
 ### `mq/` — Message Queue
@@ -160,7 +160,7 @@ Bento ingest pipeline (StartIngestWorker):
   → On failure: route to DLQ output (dlq.{table}), then Ack to prevent infinite retry
 
   (Insert-only pipeline. The wire format `EventMessage` carries only
-  {table_name, received_timestamp, data}; non-insert mutations
+  {table_name, scope, received_timestamp, data}; non-insert mutations
   DELETE/UPDATE/TRUNCATE/DROP/etc. must go through POST /v1/admin/query — when
   authentication is enabled, the /v1/admin/* role gate rejects non-admin
   callers at the API layer (with auth.enabled=false or auth.dev_mode=true
@@ -191,7 +191,7 @@ Client POST /v1/admin/query
     <scheme>://<host>:<httpport>/?default_format=JSON
        &date_time_output_format=iso&database=<db>
     Auth via X-ClickHouse-User / X-ClickHouse-Key headers.
-    Bound by a 30s context derived from the inbound request — client
+    Bound by a clickhouse.query_timeout context derived from the inbound request — client
     disconnect cancels the upstream call.
   → ClickHouse parses the SQL natively and decides what to do:
     → Read: returns 200 + {"meta":[...], "data":[...], "rows":N,

docs/src/content/docs/configuration.md

@@ -50,6 +50,7 @@ export WH_CONFIG=/etc/wavehouse/config.yaml
 | `clickhouse.database` | `WH_CH_DATABASE` | `default` | Database name. Tables are discovered from this database. |
 | `clickhouse.username` | `WH_CH_USERNAME` | `default` | Authentication username. |
 | `clickhouse.password` | `WH_CH_PASSWORD` | *(empty)* | Authentication password. |
+| `clickhouse.query_timeout` | `WH_CH_QUERY_TIMEOUT` | `30s` | Maximum allowed execution time for ClickHouse queries |
 
 ### Schema Discovery
 
@@ -76,7 +77,6 @@ export WH_CONFIG=/etc/wavehouse/config.yaml
 | YAML Key | Env Var | Default | Description |
 | -------- | ------- | ------- | ----------- |
 | `cache.l1_max_cost` | `WH_CACHE_L1_MAX_COST` | `67108864` | Maximum L1 cache size in bytes (~64 MB). |
-| `cache.default_ttl` | `WH_CACHE_DEFAULT_TTL` | `300` | Default cache TTL in seconds (5 minutes). |
 | `cache.timestamp_bucket_seconds` | `WH_CACHE_TIMESTAMP_BUCKET_SECONDS` | `60` | Bucket size (seconds) for time-range truncation in structured queries. Improves cache hit rate by normalizing timestamps. |
 
 ### Authentication
@@ -167,7 +167,6 @@ dedupe:
 
 cache:
   l1_max_cost: 67108864
-  default_ttl: 300
   timestamp_bucket_seconds: 60
 
 auth:

docs/src/content/docs/sdk.md

@@ -260,10 +260,6 @@ clicks.select('page').timeRange('received_timestamp', '1h')
 clicks.select('page').timeRange('received_timestamp', '2026-01-01T00:00:00Z', '2026-02-01T00:00:00Z')
 ```
 
-#### `.cacheTTL(seconds)`
-
-Override the server's default cache TTL for this query.
-
 ```ts
 clicks.select('page').count().cacheTTL(300) // cache for 5 minutes
 ```

internal/api/ingest.go

@@ -128,9 +128,13 @@ func (h *IngestHandler) Handle(w http.ResponseWriter, r *http.Request) {
 		}
 	}
 
+	// TODO: set a scope (e.g., "org_id:123") – but scope requires us to know if a table is globally shared or scoped fully by roles/org/tenant. Currently we have no real way to set this, so scopes will be empty.
+	scope := ""
+
 	now := time.Now().UTC()
 	evt := ingest.EventMessage{
 		TableName:         table,
+		Scope:             scope,
 		ReceivedTimestamp: now.Format(time.RFC3339Nano),
 		Data:              data,
 	}
@@ -141,7 +145,12 @@ func (h *IngestHandler) Handle(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	subject := "ingest." + query.EncodeTable(table)
+	subject := "ingest." + query.SafeEncodeNATS(table)
+	if scope != "" {
+		subject += "." + query.SafeEncodeNATS(scope)
+	}
+
+	slog.DebugContext(ctx, "publishing event to NATS", "subject", subject, "table", table, "scope", scope)
 	if err := h.Publisher.Publish(ctx, subject, payload); err != nil {
 		if strings.Contains(err.Error(), "maximum bytes exceeded") {
 			slog.WarnContext(ctx, "nats maximum bytes exceeded", "subject", subject)

internal/api/pipes.go

@@ -9,21 +9,22 @@ import (
 	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
 	"github.com/Wave-RF/WaveHouse/internal/cache"
 	"github.com/Wave-RF/WaveHouse/internal/pipes"
+	"github.com/Wave-RF/WaveHouse/internal/query"
 	"github.com/go-chi/chi/v5"
 	"golang.org/x/sync/singleflight"
 )
 
 // PipesHandler handles named query pipe endpoints.
 type PipesHandler struct {
-	Store      *pipes.Store
-	CHConn     driver.Conn
-	Cache      *cache.TieredCache
-	DefaultTTL time.Duration
-	sf         singleflight.Group
+	Store           *pipes.Store
+	CHConn          driver.Conn
+	Cache           cache.Cache
+	sf              singleflight.Group
+	maxQueryTimeout time.Duration
 }
 
-func NewPipesHandler(store *pipes.Store, conn driver.Conn, c *cache.TieredCache, defaultTTL time.Duration) *PipesHandler {
-	return &PipesHandler{Store: store, CHConn: conn, Cache: c, DefaultTTL: defaultTTL}
+func NewPipesHandler(store *pipes.Store, conn driver.Conn, c cache.Cache, queryTimeout time.Duration) *PipesHandler {
+	return &PipesHandler{Store: store, CHConn: conn, Cache: c, maxQueryTimeout: queryTimeout}
 }
 
 // List returns all named queries (admin endpoint).
@@ -123,10 +124,16 @@ func (h *PipesHandler) Execute(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	// TODO: scope impl
+	scope := ""
+	safePipeName := query.SafeEncodeNATS(name)
+
+	// TODO: current pipe impl doesn't have a list of tables/scopes, so bento ingest cannot invalidate it
+
 	// Cache.
 	cacheKey := queryCacheKey(sql, params)
 	if h.Cache != nil {
-		if data, _, err := h.Cache.Get(r.Context(), cacheKey); err == nil && data != nil {
+		if data, _, err := h.Cache.Get(r.Context(), cacheKey, safePipeName, scope); err == nil && data != nil {
 			w.Header().Set("Content-Type", "application/json")
 			w.Header().Set("X-Cache", "HIT")
 			_, _ = w.Write(data)
@@ -136,21 +143,28 @@ func (h *PipesHandler) Execute(w http.ResponseWriter, r *http.Request) {
 
 	// Execute with singleflight.
 	v, err, _ := h.sf.Do(cacheKey, func() (interface{}, error) {
-		queryCtx, cancel := context.WithTimeout(r.Context(), 30*time.Second)
+		queryCtx, cancel := context.WithTimeout(r.Context(), h.maxQueryTimeout)
 		defer cancel()
 
+		start := time.Now()
+
 		rows, err := executeCHQuery(queryCtx, h.CHConn, sql, params)
+		queryDuration := time.Since(start)
 		if err != nil {
+			// TODO: depending on the error, we may actually want to cache it
 			return nil, err
 		}
 
 		data, err := json.Marshal(rows)
 		if err != nil {
+			// TODO: eventually we want CSV support etc
 			return nil, err
 		}
 
+		ttl := cache.QueryTimeToTTL(queryDuration)
+
 		if h.Cache != nil {
-			_ = h.Cache.Set(r.Context(), cacheKey, data, h.DefaultTTL)
+			_ = h.Cache.Set(r.Context(), cacheKey, safePipeName, scope, data, ttl)
 		}
 		return data, nil
 	})

internal/api/query.go

@@ -49,6 +49,8 @@ type QueryHandler struct {
 	Username string
 	Password string
 	Database string
+	// maxQueryTimeout optionally ends an in-flight query when time is reached.
+	maxQueryTimeout time.Duration
 	// maxResponseBytes optionally overrides the default upstream response
 	// buffer cap (maxCHResponseBytes). When 0, the default applies. Exists
 	// so same-package tests can pin the cap-overflow path without
@@ -87,12 +89,12 @@ const (
 // 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 30s deadline
+// 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 30s).
-func NewQueryHandler(endpoint, username, password, database string) *QueryHandler {
+// inbound context longer than queryTimeout).
+func NewQueryHandler(endpoint, username, password, database string, queryTimeout time.Duration) *QueryHandler {
 	return &QueryHandler{
 		HTTPClient: &http.Client{
 			// ClickHouse's HTTP interface doesn't 3xx in normal operation,
@@ -105,10 +107,11 @@ func NewQueryHandler(endpoint, username, password, database string) *QueryHandle
 				return http.ErrUseLastResponse
 			},
 		},
-		Endpoint: endpoint,
-		Username: username,
-		Password: password,
-		Database: database,
+		Endpoint:        endpoint,
+		Username:        username,
+		Password:        password,
+		Database:        database,
+		maxQueryTimeout: queryTimeout,
 	}
 }
 
@@ -190,7 +193,7 @@ func (h *QueryHandler) Handle(w http.ResponseWriter, r *http.Request) {
 
 	// Bound the upstream call with a deadline derived from the inbound
 	// request context — client disconnect cancels the ClickHouse call too.
-	ctx, cancel := context.WithTimeout(r.Context(), 30*time.Second)
+	ctx, cancel := context.WithTimeout(r.Context(), h.maxQueryTimeout)
 	defer cancel()
 
 	// Defensive: NewQueryHandler always sets HTTPClient, but a zero-value