feat(cache): implement per-table tagging and raw SQL bypass

CHANGELOG.md

@@ -6,6 +6,18 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## Unreleased
+### Added
+- **Per-table cache invalidation via Bento** (`internal/cache/`, `internal/ingest/bento.go`): the ingest worker now triggers a targeted cache purge after successful ClickHouse batch flushes. This ensures "Read-Your-Writes" consistency within the standalone deployment without requiring a complex WAL-merging query engine.
+- **Cache Tagging system** (`internal/cache/local.go`, `internal/cache/tiered.go`): the `Cache` interface now supports `Set(..., tags []string)` and `InvalidateByTags(ctx, tags)`. `LocalCache` uses a double-indexed map (`tagsMap` and `keyTags`) to track which cache keys are associated with which tables, with automatic cleanup of the tag index on eviction to prevent memory leaks.
+- **Safe SQL Tag Extraction** (`internal/api/query.go`, `internal/api/pipes.go`): introduced `extractCacheTags()` helper that parses SQL for `FROM`, `JOIN`, and `INTO` clauses to automatically tag cache entries. Includes regex-based validation of identifiers to prevent cache-poisoning via malformed SQL.
+
+### Fixed
+- **Race condition in cache invalidation** (`internal/cache/local.go`): replaced `sync.Map` with a standard map guarded by a `sync.RWMutex` to resolve a TOCTOU (Time-of-Check to Time-of-Use) race where a concurrent `Set` could orphan a new cache entry during an active `Invalidate` sweep.
+- **Coarse test mocks** (`internal/testutil/mocks.go`, `internal/cache/tiered_test.go`): updated test mocks to correctly handle tag intersection logic, ensuring that invalidating table "A" does not accidentally wipe cached results for table "B".
+
+### Changed
+- **Breaking: Raw SQL Cache Bypass** (`internal/api/query.go`, `docs/api.md`): the `/v1/query` endpoint now strictly bypasses the cache (`X-Cache: BYPASS`). This provides a safety hatch for administrative and E2E testing workflows that require absolute real-time consistency.
+- **Breaking:** The `Cache` interface methods `Set` and `InvalidateByPrefix` (now `InvalidateByTags`) have updated signatures to support the tagging architecture.
 
 ### Removed
 

cmd/wavehouse/main.go

@@ -248,7 +248,12 @@ func run() int {
 		logger.Error("cache init", "error", err)
 		return 1
 	}
-	tiered := cache.NewTiered(l1, nil)
+	tiered, err := cache.NewTiered(l1, nil)
+	if err != nil {
+		logger.Error("tiered cache init", "error", err)
+		return 1
+	}
+
 	defer func() { _ = tiered.Close() }()
 
 	// Policy store (NATS KV + optional file bootstrap).
@@ -281,6 +286,7 @@ func run() int {
 		ctx,
 		embeddedMQ.NatsConn(),
 		chConn,
+		tiered,
 		cfg.ClickHouse.Addr,
 		cfg.ClickHouse.HTTPPort, // Uses 8123 by default
 		cfg.ClickHouse.HTTPScheme,

docs/src/content/docs/api.md

@@ -209,8 +209,9 @@ Executes a SQL query directly against ClickHouse. Results are cached in-process
 
 The response includes a cache header:
 
-- `X-Cache: HIT` — served from cache
-- `X-Cache: MISS` — fetched from ClickHouse
+- `X-Cache: BYPASS` — the query bypassed the cache to ensure consistency (standard for raw SQL). Additionally, mutation statements (`INSERT`, `UPDATE`, `DELETE`, `TRUNCATE`, `DROP`, `ALTER`, `REPLACE`) trigger [`InvalidateByTags`](../architecture.md) for the affected tables, clearing any cached results from `/v1/tables/{table}/query` and `/v1/pipes/{name}` for those tables.
+- `X-Cache: HIT` — served from cache (Standard for Pipes and Structured Queries)
+- `X-Cache: MISS` — fetched from ClickHouse and subsequently cached
 
 **Error responses:**
 

docs/src/content/docs/architecture.md

@@ -153,13 +153,18 @@ Client POST /v1/ingest/{table}
 
 Bento ingest pipeline (StartIngestWorker):
   ← JetStream pull consumer (buffer-consumer) on ingest.>
-  → Validate table name against safeIdentifierRe
+  → Validate table name against query.SafeIdentifierRe (internal/query/sqlutil.go)
   → Batch events per table, bulk INSERT to ClickHouse
   → Delete actions handled inline (chConn.Exec)
     → On success: DoubleAck and continue
     → On failure: route original envelope to dlq.{table} and DoubleAck (Phase 1: all delete errors treated as permanent to prevent infinite Nak loops)
-  → On success: DoubleAck messages
-  → On failure: route to DLQ output (dlq.{table}), then Ack to prevent infinite retry
+  → On success:
+    → Trigger Cache.InvalidateByTags([table_name])
+    → (Purges all L1/L2 entries associated with that table)
+    → DoubleAck messages to NATS
+  → On failure:
+    → Route batch to DLQ output (subject: dlq.{table})
+    → Ack to NATS to prevent infinite retry loops
 
 Active Sweeper (async goroutine, every 60s):
   → Read buffer consumer's AckFloor (highest contiguous ACKed seq)
@@ -171,14 +176,22 @@ Active Sweeper (async goroutine, every 60s):
 ### Query Path
 
 ```text
-Client POST /v1/query
+Client POST /v1/query (Raw SQL)
+  → Optional JWT auth middleware
+  → Execute query directly on ClickHouse (Cache Bypass)
+  → If SQL is a mutation (INSERT/UPDATE/DELETE/TRUNCATE/DROP/ALTER/REPLACE):
+    → Extract table names
+    → Trigger InvalidateByTags across Cache tiers
+  → Return result (X-Cache: BYPASS header)
+
+Client POST /v1/tables/{table}/query (Structured) or /v1/pipes/{name}
   → Optional JWT auth middleware
   → Check tiered cache (L1 → L2)
   → Cache HIT: return cached result (X-Cache: HIT header)
   → Cache MISS:
-    → Execute query directly on ClickHouse
-    → Convert UUID/DateTime types to strings
-    → Store result in L1 + L2
+    → Execute query on ClickHouse
+    → Extract table tags from SQL
+    → Store result in L1 + L2 associated with table tags
     → Return result (X-Cache: MISS header)
 ```
 

internal/api/pipes.go

@@ -3,6 +3,7 @@ package api
 import (
 	"context"
 	"encoding/json"
+	"log/slog"
 	"net/http"
 	"time"
 
@@ -17,12 +18,12 @@ import (
 type PipesHandler struct {
 	Store      *pipes.Store
 	CHConn     driver.Conn
-	Cache      *cache.TieredCache
+	Cache      cache.Cache
 	DefaultTTL time.Duration
 	sf         singleflight.Group
 }
 
-func NewPipesHandler(store *pipes.Store, conn driver.Conn, c *cache.TieredCache, defaultTTL time.Duration) *PipesHandler {
+func NewPipesHandler(store *pipes.Store, conn driver.Conn, c cache.Cache, defaultTTL time.Duration) *PipesHandler {
 	return &PipesHandler{Store: store, CHConn: conn, Cache: c, DefaultTTL: defaultTTL}
 }
 
@@ -134,7 +135,6 @@ 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)
 		defer cancel()
@@ -151,7 +151,20 @@ func (h *PipesHandler) Execute(w http.ResponseWriter, r *http.Request) {
 		}
 
 		if h.Cache != nil {
-			_ = h.Cache.Set(r.Context(), cacheKey, data, h.DefaultTTL)
+			cleanedSQL := cleanSQLForTags(sql)
+			tags := extractCacheTagsFromCleaned(cleanedSQL)
+
+			// Detach from r.Context() so a cancelled originating request doesn't
+			// abort the cache populate — singleflight followers may already have
+			// received the result, and the next request should be served warm.
+			setCtx, setCancel := context.WithTimeout(context.Background(), 5*time.Second)
+			defer setCancel()
+			if err := h.Cache.Set(setCtx, cacheKey, data, h.DefaultTTL, tags); err != nil {
+				slog.WarnContext(r.Context(), "cache set failed for pipe execution",
+					"key", cacheKey,
+					"error", err,
+				)
+			}
 		}
 		return data, nil
 	})

internal/api/pipes_test.go

@@ -7,8 +7,10 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"testing"
+	"time"
 
 	"github.com/Wave-RF/WaveHouse/internal/pipes"
+	"github.com/Wave-RF/WaveHouse/internal/testutil"
 	"github.com/go-chi/chi/v5"
 	"github.com/golang-jwt/jwt/v5"
 	"github.com/stretchr/testify/assert"
@@ -327,3 +329,31 @@ func TestPipesHandler_Execute_NoRolesAllowsAll(t *testing.T) {
 	assert.NotEqual(t, http.StatusForbidden, w.Code)
 	assert.NotEqual(t, http.StatusNotFound, w.Code)
 }
+
+func TestPipesHandler_Execute_CacheTags(t *testing.T) {
+	t.Parallel()
+
+	mockCache := testutil.NewMockCache()
+	mockConn := &testutil.MockConn{}
+
+	pipe := &pipes.NamedQuery{
+		Name: "active_users",
+		SQL:  "SELECT * FROM users WHERE active = 1",
+	}
+	store := pipes.NewMemoryStore(pipe)
+	h := NewPipesHandler(store, mockConn, mockCache, time.Minute)
+
+	r := pipesRequest(t, http.MethodPost, "/v1/pipes/active_users/execute", "active_users", nil)
+	w := httptest.NewRecorder()
+	h.Execute(w, r)
+
+	require.Equal(t, http.StatusOK, w.Code)
+
+	// Reconstruct the key the handler computed so we can probe the mock.
+	sql, params, err := pipes.BindParams(pipe, map[string]any{})
+	require.NoError(t, err)
+	key := queryCacheKey(sql, params)
+
+	assert.Equal(t, []string{"users"}, mockCache.TagsForKey(key),
+		"Execute must call Cache.Set with the table tags extracted from the pipe SQL")
+}