feat(streaming): query cache

api/v3/handlers/meters/query.go

@@ -59,6 +59,8 @@ func (h *handler) QueryMeter() QueryMeterHandler {
 				return QueryMeterResponse{}, err
 			}
 
+			params.Cachable = true
+
 			rows, err := h.streaming.QueryMeter(ctx, req.Namespace, m, params)
 			if err != nil {
 				return QueryMeterResponse{}, err

api/v3/handlers/meters/query_csv.go

@@ -72,6 +72,8 @@ func (h *handler) QueryMeterCSV() QueryMeterCSVHandler {
 				return nil, err
 			}
 
+			params.Cachable = true
+
 			rows, err := h.streaming.QueryMeter(ctx, req.Namespace, m, params)
 			if err != nil {
 				return nil, err

app/common/streaming.go

@@ -7,13 +7,16 @@ import (
 
 	"github.com/ClickHouse/clickhouse-go/v2"
 	"github.com/google/wire"
+	"go.opentelemetry.io/otel/metric"
+	"go.opentelemetry.io/otel/trace"
 
 	"github.com/openmeterio/openmeter/app/config"
 	"github.com/openmeterio/openmeter/openmeter/namespace"
 	"github.com/openmeterio/openmeter/openmeter/progressmanager"
 	"github.com/openmeterio/openmeter/openmeter/streaming"
 	clickhouseconnector "github.com/openmeterio/openmeter/openmeter/streaming/clickhouse"
 	streamingretry "github.com/openmeterio/openmeter/openmeter/streaming/retry"
+	"github.com/openmeterio/openmeter/pkg/featuregate"
 )
 
 var Streaming = wire.NewSet(
@@ -27,6 +30,9 @@ func NewStreamingConnector(
 	logger *slog.Logger,
 	progressmanager progressmanager.Service,
 	namespaceManager *namespace.Manager,
+	featureGate *featuregate.FeatureGateChecker,
+	meter metric.Meter,
+	tracer trace.Tracer,
 ) (streaming.Connector, error) {
 	var connector streaming.Connector
 	var err error
@@ -43,6 +49,14 @@ func NewStreamingConnector(
 		EnablePrewhere:         conf.EnablePrewhere,
 		EnableDecimalPrecision: conf.EnableDecimalPrecision,
 		ProgressManager:        progressmanager,
+
+		QueryCacheEnabled:                     conf.QueryCache.Enabled,
+		QueryCacheMinimumCacheableQueryPeriod: conf.QueryCache.MinimumCacheableQueryPeriod,
+		QueryCacheMinimumCacheableUsageAge:    conf.QueryCache.MinimumCacheableUsageAge,
+		QueryCacheParityCheckSampleRate:       conf.QueryCache.ParityCheckSampleRate,
+		Meter:                                 meter,
+		Tracer:                                tracer,
+		FeatureGate:                           featureGate,
 	})
 	if err != nil {
 		return nil, fmt.Errorf("init clickhouse connector: %w", err)

app/config/aggregation.go

@@ -41,6 +41,9 @@ type AggregationConfiguration struct {
 	// When enabled, values are calculated using Decimal128 instead of Float64, providing
 	// higher precision for financial calculations at the cost of some performance.
 	EnableDecimalPrecision bool
+
+	// QueryCache configures the optional meter query-result cache.
+	QueryCache AggregationQueryCacheConfiguration
 }
 
 // Validate validates the configuration.
@@ -57,9 +60,69 @@ func (c AggregationConfiguration) Validate() error {
 		return errors.New("async insert wait is set but async insert is not")
 	}
 
+	if err := c.QueryCache.Validate(); err != nil {
+		return fmt.Errorf("query cache: %w", err)
+	}
+
+	if c.QueryCache.Enabled && !c.EnableDecimalPrecision {
+		return errors.New("query cache requires enableDecimalPrecision to be true")
+	}
+
+	// The cache's late-event invalidation must run after the event is visible
+	// to SELECTs; async insert without wait acks before the buffer flush, which
+	// would let a read cache rollups missing the event with no invalidation
+	// ever coming for it.
+	if c.QueryCache.Enabled && c.AsyncInsert && !c.AsyncInsertWait {
+		return errors.New("query cache requires asyncInsertWait when asyncInsert is enabled")
+	}
+
 	return nil
 }
 
+// AggregationQueryCacheConfiguration configures the optional meter query-result
+// cache: an hourly pre-aggregated rollup table that serves the settled history
+// of cacheable queries while the fresh tail is scanned live. Off by default.
+type AggregationQueryCacheConfiguration struct {
+	// Enabled turns the cache on. When false the live query path is unchanged and
+	// no cache table is created.
+	Enabled bool
+	// MinimumCacheableQueryPeriod is the minimum queried span for a query to be
+	// worth caching.
+	MinimumCacheableQueryPeriod time.Duration
+	// MinimumCacheableUsageAge is the freshness horizon: only windows entirely
+	// older than now minus this age are cached; the fresher tail is served live.
+	MinimumCacheableUsageAge time.Duration
+	// ParityCheckSampleRate (0..1) samples cache-served queries for shadow
+	// verification against the live query. A detected mismatch is logged, exposed
+	// as streaming.query_cache.parity_checks{outcome=mismatch}, and self-heals by
+	// invalidating the namespace's cache. Each sampled query re-runs the full
+	// live query, so keep the rate low (e.g. 0.01). 0 disables.
+	ParityCheckSampleRate float64
+}
+
+// Validate validates the configuration.
+func (c AggregationQueryCacheConfiguration) Validate() error {
+	if !c.Enabled {
+		return nil
+	}
+
+	var errs []error
+
+	if c.MinimumCacheableQueryPeriod <= 0 {
+		errs = append(errs, errors.New("minimum cacheable query period is required"))
+	}
+
+	if c.MinimumCacheableUsageAge <= 0 {
+		errs = append(errs, errors.New("minimum cacheable usage age is required"))
+	}
+
+	if c.ParityCheckSampleRate < 0 || c.ParityCheckSampleRate > 1 {
+		errs = append(errs, errors.New("parity check sample rate must be between 0 and 1"))
+	}
+
+	return errors.Join(errs...)
+}
+
 // ClickHouseAggregationConfiguration is the configuration for the ClickHouse aggregation engine
 type ClickHouseAggregationConfiguration struct {
 	Address  string
@@ -225,4 +288,10 @@ func ConfigureAggregation(v *viper.Viper) {
 
 	// Decimal precision
 	v.SetDefault("aggregation.enableDecimalPrecision", false)
+
+	// Query cache (off by default)
+	v.SetDefault("aggregation.queryCache.enabled", false)
+	v.SetDefault("aggregation.queryCache.minimumCacheableQueryPeriod", "168h")
+	v.SetDefault("aggregation.queryCache.minimumCacheableUsageAge", "24h")
+	v.SetDefault("aggregation.queryCache.parityCheckSampleRate", 0.0)
 }

app/config/config_test.go

@@ -160,6 +160,11 @@ func TestComplete(t *testing.T) {
 			EventsTableName: "om_events",
 			AsyncInsert:     false,
 			AsyncInsertWait: false,
+			QueryCache: AggregationQueryCacheConfiguration{
+				Enabled:                     false,
+				MinimumCacheableQueryPeriod: 72 * time.Hour,
+				MinimumCacheableUsageAge:    24 * time.Hour,
+			},
 		},
 		Entitlements: EntitlementsConfiguration{
 			GracePeriod: datetime.ISODurationString("P1D"),
@@ -494,3 +499,39 @@ func TestComplete(t *testing.T) {
 
 	assert.Equal(t, expected, actual)
 }
+
+// TestAggregationQueryCacheValidation pins the cross-field rule: the query
+// cache stores exact Decimal128 rollups, so enabling it without decimal
+// precision would ship a silently dead cache (table created, invalidation
+// running, zero queries served from it). Validate must reject the combination.
+func TestAggregationQueryCacheValidation(t *testing.T) {
+	base := AggregationConfiguration{
+		ClickHouse: ClickHouseAggregationConfiguration{
+			Address:         "127.0.0.1:9440",
+			DialTimeout:     10 * time.Second,
+			MaxOpenConns:    5,
+			MaxIdleConns:    5,
+			ConnMaxLifetime: 10 * time.Minute,
+			BlockBufferSize: 10,
+		},
+		EventsTableName: "om_events",
+		QueryCache: AggregationQueryCacheConfiguration{
+			Enabled:                     true,
+			MinimumCacheableQueryPeriod: 72 * time.Hour,
+			MinimumCacheableUsageAge:    24 * time.Hour,
+		},
+	}
+
+	withoutDecimal := base
+	withoutDecimal.EnableDecimalPrecision = false
+	assert.ErrorContains(t, withoutDecimal.Validate(), "enableDecimalPrecision")
+
+	withDecimal := base
+	withDecimal.EnableDecimalPrecision = true
+	assert.NoError(t, withDecimal.Validate())
+
+	disabled := base
+	disabled.QueryCache.Enabled = false
+	disabled.EnableDecimalPrecision = false
+	assert.NoError(t, disabled.Validate(), "the rule only applies when the cache is enabled")
+}

app/config/testdata/complete.yaml

@@ -66,6 +66,11 @@ aggregation:
     retry:
       enabled: true
 
+  queryCache:
+    enabled: false
+    minimumCacheableQueryPeriod: 72h
+    minimumCacheableUsageAge: 24h
+
 sink:
   groupId: openmeter-sink-worker
   minCommitCount: 500

cmd/balance-worker/wire.go

@@ -36,6 +36,7 @@ func initializeApplication(ctx context.Context, conf config.Configuration) (Appl
 		common.Customer,
 		common.Database,
 		common.Feature,
+		common.FeatureGateChecker,
 		common.Framework,
 		common.Meter,
 		common.Namespace,

cmd/notification-service/wire.go

@@ -27,15 +27,15 @@ type Application struct {
 	common.GlobalInitializer
 	common.Migrator
 
-	BrokerOptions      watermillkafka.BrokerOptions
-	EventPublisher     eventbus.Publisher
-	EntClient          *db.Client
-	FeatureConnector   feature.FeatureConnector
-	Logger             *slog.Logger
-	MessagePublisher   message.Publisher
-	Meter              metric.Meter
-	Tracer             trace.Tracer
-	Metadata           common.Metadata
+	BrokerOptions           watermillkafka.BrokerOptions
+	EventPublisher          eventbus.Publisher
+	EntClient               *db.Client
+	FeatureConnector        feature.FeatureConnector
+	Logger                  *slog.Logger
+	MessagePublisher        message.Publisher
+	Meter                   metric.Meter
+	Tracer                  trace.Tracer
+	Metadata                common.Metadata
 	MeterService            meter.Service
 	Notification            notification.Service
 	RuntimeMetricsCollector common.RuntimeMetricsCollector
@@ -50,6 +50,7 @@ func initializeApplication(ctx context.Context, conf config.Configuration) (Appl
 		common.Config,
 		common.Database,
 		common.Feature,
+		common.FeatureGateChecker,
 		common.Framework,
 		common.Meter,
 		common.Namespace,