internal/repository/clickhouse/aggregators.go
Rama main · réplica de lectura del repositorio original.
// Package clickhouse provides ClickHouse-specific aggregators for usage data.
//
// Custom Monthly Billing Periods
// ==============================
//
// This package supports custom monthly billing periods through the BillingAnchor parameter.
// This allows usage data to be aggregated by custom monthly periods (e.g., 5th to 5th of each month)
// instead of standard calendar months (1st to 1st of each month).
//
// How it works:
// 1. When WindowSize = "MONTH" and BillingAnchor is provided, events are grouped by custom monthly periods
// 2. The BillingAnchor timestamp defines the reference day for monthly periods (time is ignored)
// 3. Day-level granularity is used for simplicity and predictability
// 4. All other window sizes (DAY, HOUR, WEEK, etc.) ignore BillingAnchor and use standard windows
//
// Example:
// BillingAnchor = 2024-03-05 (any time on March 5th)
// - March period: 2024-03-05 to 2024-04-05
// - April period: 2024-04-05 to 2024-05-05
//
// Use cases:
// - Subscription billing that doesn't align with calendar months
// - Customer-specific billing cycles (e.g., signed up on 15th)
// - Multi-tenant systems with different billing anchor dates
// - Custom business cycles (fiscal months, quarterly periods)
//
// Implementation:
// The custom monthly logic generates ClickHouse expressions that:
// 1. Shift timestamps by the day offset from the billing anchor
// 2. Apply toStartOfMonth() to get calendar month boundaries
// 3. Shift back by the same day offset to create custom monthly periods
//
// This ensures that events are correctly grouped into the appropriate billing periods
// using day-level granularity for better predictability and user understanding.
//
// Query Parameterization
// =======================
//
// All queries built in this file use `?` positional placeholders for any value that
// originates from user/API input (filter property names/values, customer IDs, event
// name, timestamps). Each query-building function returns (query string, []interface{}
// args) — the args slice must be passed to the ClickHouse driver's Query(ctx, query,
// args...) call in the same order the placeholders appear in the concatenated query
// string, since ClickHouse binds positionally. This mirrors the pattern already used in
// meter_usage_query_builder.go.
//
// Two inputs cannot be bound and are instead validated then interpolated, because they
// are used as SQL identifiers rather than values:
// - timezone: validated by normalizeCHTimezone (constrained to resolvable IANA names
// via time.LoadLocation) before it is interpolated into the window expressions.
// - group-by property: validated by validateGroupByProperty (strict allow-list) before
// it is interpolated as the JSONExtractString key.
package clickhouse
import (
"context"
"fmt"
"strings"
"time"
"github.com/flexprice/flexprice/internal/domain/events"
"github.com/flexprice/flexprice/internal/types"
"github.com/samber/lo"
"github.com/shopspring/decimal"
)
func GetAggregator(aggregationType types.AggregationType) events.Aggregator {
switch aggregationType {
case types.AggregationCount:
return &CountAggregator{}
case types.AggregationSum:
return &SumAggregator{}
case types.AggregationAvg:
return &AvgAggregator{}
case types.AggregationCountUnique:
return &CountUniqueAggregator{}
case types.AggregationLatest:
return &LatestAggregator{}
case types.AggregationSumWithMultiplier:
return &SumWithMultiAggregator{}
case types.AggregationMax:
return &MaxAggregator{}
case types.AggregationWeightedSum:
return &WeightedSumAggregator{}
}
return nil
}
func getDeduplicationKey() string {
return "id"
}
// buildUsageEventCustomerFilters returns PREWHERE fragments (with `?` placeholders) for
// external_customer_id and FlexPrice customer_id, plus the args to bind against them, in
// the order the two returned fragments are concatenated into the final query. Params may
// be nil. External merges ExternalCustomerID and ExternalCustomerIDs (deduped); internal
// uses CustomerID only.
func buildUsageEventCustomerFilters(params *events.UsageParams) (externalCustomerFilter string, customerFilter string, args []interface{}) {
if params == nil {
return "", "", nil
}
extIDs := make([]string, 0)
if params.ExternalCustomerID != "" {
extIDs = append(extIDs, params.ExternalCustomerID)
}
if len(params.ExternalCustomerIDs) > 0 {
extIDs = append(extIDs, params.ExternalCustomerIDs...)
}
extUnique := lo.Uniq(extIDs)
switch len(extUnique) {
case 0:
externalCustomerFilter = ""
case 1:
externalCustomerFilter = "AND external_customer_id = ?"
args = append(args, extUnique[0])
default:
placeholders := make([]string, len(extUnique))
for i := range extUnique {
placeholders[i] = "?"
}
externalCustomerFilter = fmt.Sprintf("AND external_customer_id IN (%s)", strings.Join(placeholders, ", "))
for _, id := range extUnique {
args = append(args, id)
}
}
if params.CustomerID != "" {
customerFilter = "AND customer_id = ?"
args = append(args, params.CustomerID)
}
return externalCustomerFilter, customerFilter, args
}
// normalizeCHTimezone returns a valid IANA timezone name for ClickHouse
// toStartOf* functions, defaulting to UTC for empty, "UTC", or unresolvable
// values. Passing an invalid name straight to ClickHouse would error the query,
// so this is the single guard shared by both window formatters. Since the return
// value is validated against time.LoadLocation, it is safe to interpolate directly.
func normalizeCHTimezone(tz string) string {
if tz == "" || tz == types.DefaultTimezone {
return types.DefaultTimezone
}
if _, err := time.LoadLocation(tz); err != nil {
return types.DefaultTimezone
}
return tz
}
// formatWindowSize returns the ClickHouse expression that buckets `timestamp`
// into the given window in the given timezone. "UTC" is a valid ClickHouse
// timezone argument and yields identical results to omitting it, so the tz is
// always passed — there is no UTC special-casing. The tz value has already been
// validated by normalizeCHTimezone (constrained to resolvable IANA names via
// time.LoadLocation), so it is safe to interpolate directly here rather than bind
// as a query arg — these window expressions are used inside larger fmt.Sprintf
// templates that are themselves fully parameterized for user-controlled values.
func formatWindowSize(windowSize types.WindowSize, tz string) string {
if windowSize == "" {
return ""
}
// Minute buckets are timezone-invariant (all real offsets are whole minutes).
if windowSize == types.WindowSizeMinute {
return "toStartOfMinute(timestamp)"
}
tz = normalizeCHTimezone(tz)
switch windowSize {
case types.WindowSize15Min:
return fmt.Sprintf("toStartOfInterval(timestamp, INTERVAL 15 MINUTE, '%s')", tz)
case types.WindowSize30Min:
return fmt.Sprintf("toStartOfInterval(timestamp, INTERVAL 30 MINUTE, '%s')", tz)
case types.WindowSize3Hour:
return fmt.Sprintf("toStartOfInterval(timestamp, INTERVAL 3 HOUR, '%s')", tz)
case types.WindowSize6Hour:
return fmt.Sprintf("toStartOfInterval(timestamp, INTERVAL 6 HOUR, '%s')", tz)
case types.WindowSize12Hour:
return fmt.Sprintf("toStartOfInterval(timestamp, INTERVAL 12 HOUR, '%s')", tz)
case types.WindowSizeHour:
return fmt.Sprintf("toStartOfHour(timestamp, '%s')", tz)
case types.WindowSizeWeek:
return fmt.Sprintf("toStartOfWeek(timestamp, 0, '%s')", tz)
case types.WindowSizeMonth:
return fmt.Sprintf("toStartOfMonth(timestamp, '%s')", tz)
case types.WindowSizeDay:
return fmt.Sprintf("toStartOfDay(timestamp, '%s')", tz)
default:
return fmt.Sprintf("toStartOfDay(timestamp, '%s')", tz)
}
}
// formatWindowSizeWithBillingAnchor formats window size with custom billing anchor for monthly periods.
//
// This function handles custom billing periods for monthly aggregations, allowing events to be grouped
// by custom monthly periods (e.g., 5th to 5th of each month) instead of calendar months (1st to 1st).
//
// Parameters:
// - windowSize: The aggregation window size (MONTH, DAY, HOUR, etc.)
// - billingAnchor: The reference timestamp for custom monthly periods (only used for MONTH window size)
//
// Behavior by window size:
// - MONTH + billingAnchor: Creates custom monthly periods using day-level granularity
// - MONTH + nil: Falls back to standard calendar months (toStartOfMonth)
// - DAY/HOUR/WEEK/etc: Ignores billing anchor, uses standard window functions
//
// Example: billingAnchor = 2024-03-05 (any time on March 5th)
// - Events from March 5 to April 5 → March billing period
// - Events from April 5 to May 5 → April billing period
//
// The generated ClickHouse expression shifts timestamps by the day offset,
// applies toStartOfMonth(), then shifts back to create the correct billing periods.
// This uses day-level granularity for simplicity and predictability.
func formatWindowSizeWithBillingAnchor(windowSize types.WindowSize, billingAnchor *time.Time, tz string) string {
if windowSize == types.WindowSizeMonth && billingAnchor != nil {
// Extract only the day component from billing anchor for simplicity
anchorDay := billingAnchor.Day()
tz = normalizeCHTimezone(tz)
// Custom monthly window anchored on anchorDay, computed in the customer's
// timezone. Shift by the day offset, snap to month start, shift back.
// "UTC" is a valid tz argument and reproduces the legacy (UTC) result.
return fmt.Sprintf("addDays(toStartOfMonth(addDays(toTimezone(timestamp, '%s'), -%d), '%s'), %d)", tz, anchorDay-1, tz, anchorDay-1)
}
// Fall back to standard window size formatting
return formatWindowSize(windowSize, tz)
}
// buildFilterConditions builds JSONExtractString filter fragments using positional `?`
// placeholders for both the property name and its value(s) — both are user-controlled
// (request body `filters` map) and must never be interpolated into the query string.
// Returns the WHERE-fragment (e.g. "AND JSONExtractString(properties, ?) IN (?,?) AND ...")
// and the args to bind, in the exact order the placeholders appear.
func buildFilterConditions(filters map[string][]string) (string, []interface{}) {
if len(filters) == 0 {
return "", nil
}
var conditions []string
var args []interface{}
for key, values := range filters {
if len(values) == 0 {
continue
}
placeholders := make([]string, len(values))
for i := range values {
placeholders[i] = "?"
}
conditions = append(conditions, fmt.Sprintf(
"JSONExtractString(properties, ?) IN (%s)",
strings.Join(placeholders, ","),
))
args = append(args, key)
for _, v := range values {
args = append(args, v)
}
}
if len(conditions) == 0 {
return "", nil
}
return "AND " + strings.Join(conditions, " AND "), args
}
// buildTimeConditions builds the time-range WHERE fragment using `?` placeholders.
func buildTimeConditions(params *events.UsageParams) (string, []interface{}) {
conditions, args := parseTimeConditions(params)
if len(conditions) == 0 {
return "", nil
}
return "AND " + strings.Join(conditions, " AND "), args
}
// appendArgs concatenates arg groups onto base in order — a small helper to
// collapse the repeated `args = append(args, xArgs...)` chain that follows
// every query-building function's args slice construction in this file.
func appendArgs(base []interface{}, groups ...[]interface{}) []interface{} {
for _, g := range groups {
base = append(base, g...)
}
return base
}
// parseTimeConditions binds start/end time.Time values directly to `?` placeholders.
// The clickhouse-go driver's formatTime emits an explicit `'UTC'` third arg when the
// value's location doesn't match the server's, so the timestamp is compared as a UTC
// instant regardless of server timezone. Formatting to a string first and wrapping in
// toDateTime64('...', 3) omits the timezone and lets the server reinterpret the value
// in its local tz — the exact bug this replaces.
func parseTimeConditions(params *events.UsageParams) ([]string, []interface{}) {
var conditions []string
var args []interface{}
if !params.StartTime.IsZero() {
conditions = append(conditions, "timestamp >= ?")
args = append(args, params.StartTime.UTC())
}
if !params.EndTime.IsZero() {
conditions = append(conditions, "timestamp < ?")
args = append(args, params.EndTime.UTC())
}
return conditions, args
}
// SumAggregator implements sum aggregation
type SumAggregator struct{}
func (a *SumAggregator) GetQuery(ctx context.Context, params *events.UsageParams) (string, []interface{}) {
// If bucket_size is specified, use windowed aggregation
if params.BucketSize != "" {
return a.getWindowedQuery(ctx, params)
}
// Otherwise use simple SUM aggregation
return a.getNonWindowedQuery(ctx, params)
}
func (a *SumAggregator) getNonWindowedQuery(ctx context.Context, params *events.UsageParams) (string, []interface{}) {
windowSize := formatWindowSizeWithBillingAnchor(params.WindowSize, params.BillingAnchor, params.Timezone)
selectClause := ""
windowClause := ""
groupByClause := ""
windowGroupBy := ""
if windowSize != "" {
selectClause = "window_size,"
windowClause = fmt.Sprintf("%s AS window_size,", windowSize)
groupByClause = "GROUP BY window_size ORDER BY window_size"
windowGroupBy = ", window_size"
}
externalCustomerFilter, customerFilter, customerArgs := buildUsageEventCustomerFilters(params)
filterConditions, filterArgs := buildFilterConditions(params.Filters)
timeConditions, timeArgs := buildTimeConditions(params)
query := fmt.Sprintf(`
SELECT
%s sum(value) as total
FROM (
SELECT
%s anyLast(JSONExtractFloat(assumeNotNull(properties), ?)) as value
FROM events
PREWHERE tenant_id = ?
AND environment_id = ?
AND event_name = ?
%s
%s
%s
%s
GROUP BY %s %s
)
%s
`,
selectClause,
windowClause,
externalCustomerFilter,
customerFilter,
filterConditions,
timeConditions,
getDeduplicationKey(),
windowGroupBy,
groupByClause)
args := appendArgs([]interface{}{params.PropertyName, types.GetTenantID(ctx), types.GetEnvironmentID(ctx), params.EventName}, customerArgs, filterArgs, timeArgs)
return query, args
}
func (a *SumAggregator) getWindowedQuery(ctx context.Context, params *events.UsageParams) (string, []interface{}) {
bucketWindow := formatWindowSizeWithBillingAnchor(params.BucketSize, params.BillingAnchor, params.Timezone)
externalCustomerFilter, customerFilter, customerArgs := buildUsageEventCustomerFilters(params)
filterConditions, filterArgs := buildFilterConditions(params.Filters)
timeConditions, timeArgs := buildTimeConditions(params)
// Get sum values per bucket, return each bucket's sum separately
query := fmt.Sprintf(`
WITH bucket_sums AS (
SELECT
%s as bucket_start,
sum(JSONExtractFloat(assumeNotNull(properties), ?)) as bucket_sum
FROM events FINAL
PREWHERE tenant_id = ?
AND environment_id = ?
AND event_name = ?
%s
%s
%s
%s
GROUP BY bucket_start
ORDER BY bucket_start
)
SELECT
(SELECT sum(bucket_sum) FROM bucket_sums) as total,
bucket_start as timestamp,
bucket_sum as value
FROM bucket_sums
ORDER BY bucket_start
`,
bucketWindow,
externalCustomerFilter,
customerFilter,
filterConditions,
timeConditions)
args := appendArgs([]interface{}{params.PropertyName, types.GetTenantID(ctx), types.GetEnvironmentID(ctx), params.EventName}, customerArgs, filterArgs, timeArgs)
return query, args
}
func (a *SumAggregator) GetType() types.AggregationType {
return types.AggregationSum
}
// CountAggregator implements count aggregation
type CountAggregator struct{}
func (a *CountAggregator) GetQuery(ctx context.Context, params *events.UsageParams) (string, []interface{}) {
windowSize := formatWindowSizeWithBillingAnchor(params.WindowSize, params.BillingAnchor, params.Timezone)
selectClause := ""
groupByClause := ""
if windowSize != "" {
selectClause = fmt.Sprintf("%s AS window_size,", windowSize)
groupByClause = "GROUP BY window_size ORDER BY window_size"
}
externalCustomerFilter, customerFilter, customerArgs := buildUsageEventCustomerFilters(params)
filterConditions, filterArgs := buildFilterConditions(params.Filters)
timeConditions, timeArgs := buildTimeConditions(params)
query := fmt.Sprintf(`
SELECT
%s count(DISTINCT %s) as total
FROM events
PREWHERE tenant_id = ?
AND environment_id = ?
AND event_name = ?
%s
%s
%s
%s
%s
`,
selectClause,
getDeduplicationKey(),
externalCustomerFilter,
customerFilter,
filterConditions,
timeConditions,
groupByClause)
args := appendArgs([]interface{}{types.GetTenantID(ctx), types.GetEnvironmentID(ctx), params.EventName}, customerArgs, filterArgs, timeArgs)
return query, args
}
func (a *CountAggregator) GetType() types.AggregationType {
return types.AggregationCount
}
// CountUniqueAggregator implements count unique aggregation
type CountUniqueAggregator struct{}
func (a *CountUniqueAggregator) GetQuery(ctx context.Context, params *events.UsageParams) (string, []interface{}) {
windowSize := formatWindowSizeWithBillingAnchor(params.WindowSize, params.BillingAnchor, params.Timezone)
selectClause := ""
windowClause := ""
groupByClause := ""
windowGroupBy := ""
if windowSize != "" {
selectClause = "window_size,"
windowClause = fmt.Sprintf("%s AS window_size,", windowSize)
groupByClause = "GROUP BY window_size ORDER BY window_size"
windowGroupBy = ", window_size"
}
externalCustomerFilter, customerFilter, customerArgs := buildUsageEventCustomerFilters(params)
filterConditions, filterArgs := buildFilterConditions(params.Filters)
timeConditions, timeArgs := buildTimeConditions(params)
query := fmt.Sprintf(`
SELECT
%s count(DISTINCT property_value) as total
FROM (
SELECT
%s JSONExtractString(assumeNotNull(properties), ?) as property_value
FROM events
PREWHERE tenant_id = ?
AND environment_id = ?
AND event_name = ?
%s
%s
%s
%s
GROUP BY %s, property_value %s
)
%s
`,
selectClause,
windowClause,
externalCustomerFilter,
customerFilter,
filterConditions,
timeConditions,
getDeduplicationKey(),
windowGroupBy,
groupByClause)
args := appendArgs([]interface{}{params.PropertyName, types.GetTenantID(ctx), types.GetEnvironmentID(ctx), params.EventName}, customerArgs, filterArgs, timeArgs)
return query, args
}
func (a *CountUniqueAggregator) GetType() types.AggregationType {
return types.AggregationCountUnique
}
// AvgAggregator implements avg aggregation
type AvgAggregator struct{}
func (a *AvgAggregator) GetQuery(ctx context.Context, params *events.UsageParams) (string, []interface{}) {
windowSize := formatWindowSizeWithBillingAnchor(params.WindowSize, params.BillingAnchor, params.Timezone)
selectClause := ""
windowClause := ""
groupByClause := ""
windowGroupBy := ""
if windowSize != "" {
selectClause = "window_size,"
windowClause = fmt.Sprintf("%s AS window_size,", windowSize)
groupByClause = "GROUP BY window_size ORDER BY window_size"
windowGroupBy = ", window_size"
}
externalCustomerFilter, customerFilter, customerArgs := buildUsageEventCustomerFilters(params)
filterConditions, filterArgs := buildFilterConditions(params.Filters)
timeConditions, timeArgs := buildTimeConditions(params)
query := fmt.Sprintf(`
SELECT
%s avg(value) as total
FROM (
SELECT
%s anyLast(JSONExtractFloat(assumeNotNull(properties), ?)) as value
FROM events
PREWHERE tenant_id = ?
AND environment_id = ?
AND event_name = ?
%s
%s
%s
%s
GROUP BY %s %s
)
%s
`,
selectClause,
windowClause,
externalCustomerFilter,
customerFilter,
filterConditions,
timeConditions,
getDeduplicationKey(),
windowGroupBy,
groupByClause)
args := appendArgs([]interface{}{params.PropertyName, types.GetTenantID(ctx), types.GetEnvironmentID(ctx), params.EventName}, customerArgs, filterArgs, timeArgs)
return query, args
}
func (a *AvgAggregator) GetType() types.AggregationType {
return types.AggregationAvg
}
// LatestAggregator implements latest value aggregation
type LatestAggregator struct{}
func (a *LatestAggregator) GetQuery(ctx context.Context, params *events.UsageParams) (string, []interface{}) {
windowSize := formatWindowSizeWithBillingAnchor(params.WindowSize, params.BillingAnchor, params.Timezone)
windowClause := ""
groupByClause := ""
if windowSize != "" {
windowClause = fmt.Sprintf("%s AS window_size,", windowSize)
groupByClause = "GROUP BY window_size ORDER BY window_size"
}
externalCustomerFilter, customerFilter, customerArgs := buildUsageEventCustomerFilters(params)
filterConditions, filterArgs := buildFilterConditions(params.Filters)
timeConditions, timeArgs := buildTimeConditions(params)
query := fmt.Sprintf(`
SELECT
%s argMax(JSONExtractFloat(assumeNotNull(properties), ?), timestamp) as total
FROM
events
PREWHERE tenant_id = ?
AND environment_id = ?
AND event_name = ?
%s
%s
%s
%s
%s
`,
windowClause,
externalCustomerFilter,
customerFilter,
filterConditions,
timeConditions,
groupByClause)
args := appendArgs([]interface{}{params.PropertyName, types.GetTenantID(ctx), types.GetEnvironmentID(ctx), params.EventName}, customerArgs, filterArgs, timeArgs)
return query, args
}
func (a *LatestAggregator) GetType() types.AggregationType {
return types.AggregationLatest
}
// SumWithMultiAggregator implements sum with multiplier aggregation
type SumWithMultiAggregator struct{}
func (a *SumWithMultiAggregator) GetQuery(ctx context.Context, params *events.UsageParams) (string, []interface{}) {
windowSize := formatWindowSizeWithBillingAnchor(params.WindowSize, params.BillingAnchor, params.Timezone)
selectClause := ""
windowClause := ""
groupByClause := ""
windowGroupBy := ""
if windowSize != "" {
selectClause = "window_size,"
windowClause = fmt.Sprintf("%s AS window_size,", windowSize)
groupByClause = "GROUP BY window_size ORDER BY window_size"
windowGroupBy = ", window_size"
}
externalCustomerFilter, customerFilter, customerArgs := buildUsageEventCustomerFilters(params)
filterConditions, filterArgs := buildFilterConditions(params.Filters)
timeConditions, timeArgs := buildTimeConditions(params)
// Multiplier is a decimal.Decimal value computed server-side (not raw user string
// input) — safe to interpolate as a numeric literal via .String(), same as before.
multiplier := decimal.NewFromInt(1)
if params.Multiplier != nil {
multiplier = *params.Multiplier
}
query := fmt.Sprintf(`
SELECT
%s (sum(value) * %s) as total
FROM (
SELECT
%s anyLast(JSONExtractFloat(assumeNotNull(properties), ?)) as value
FROM events
PREWHERE tenant_id = ?
AND environment_id = ?
AND event_name = ?
%s
%s
%s
%s
GROUP BY %s %s
)
%s
`,
selectClause,
multiplier.String(),
windowClause,
externalCustomerFilter,
customerFilter,
filterConditions,
timeConditions,
getDeduplicationKey(),
windowGroupBy,
groupByClause)
args := appendArgs([]interface{}{params.PropertyName, types.GetTenantID(ctx), types.GetEnvironmentID(ctx), params.EventName}, customerArgs, filterArgs, timeArgs)
return query, args
}
func (a *SumWithMultiAggregator) GetType() types.AggregationType {
return types.AggregationSumWithMultiplier
}
// MaxAggregator implements max aggregation
type MaxAggregator struct{}
func (a *MaxAggregator) GetQuery(ctx context.Context, params *events.UsageParams) (string, []interface{}) {
// If bucket_size is specified, use windowed aggregation
if params.BucketSize != "" {
return a.getWindowedQuery(ctx, params)
}
// Otherwise use simple MAX aggregation
return a.getNonWindowedQuery(ctx, params)
}
func (a *MaxAggregator) getNonWindowedQuery(ctx context.Context, params *events.UsageParams) (string, []interface{}) {
windowSize := formatWindowSizeWithBillingAnchor(params.WindowSize, params.BillingAnchor, params.Timezone)
selectClause := ""
windowClause := ""
groupByClause := ""
windowGroupBy := ""
if windowSize != "" {
selectClause = "window_size,"
windowClause = fmt.Sprintf("%s AS window_size,", windowSize)
groupByClause = "GROUP BY window_size ORDER BY window_size"
windowGroupBy = ", window_size"
}
externalCustomerFilter, customerFilter, customerArgs := buildUsageEventCustomerFilters(params)
filterConditions, filterArgs := buildFilterConditions(params.Filters)
timeConditions, timeArgs := buildTimeConditions(params)
query := fmt.Sprintf(`
SELECT
%s max(value) as total
FROM (
SELECT
%s anyLast(JSONExtractFloat(assumeNotNull(properties), ?)) as value
FROM events
PREWHERE tenant_id = ?
AND environment_id = ?
AND event_name = ?
%s
%s
%s
%s
GROUP BY %s %s
)
%s
`,
selectClause,
windowClause,
externalCustomerFilter,
customerFilter,
filterConditions,
timeConditions,
getDeduplicationKey(),
windowGroupBy,
groupByClause)
args := appendArgs([]interface{}{params.PropertyName, types.GetTenantID(ctx), types.GetEnvironmentID(ctx), params.EventName}, customerArgs, filterArgs, timeArgs)
return query, args
}
func (a *MaxAggregator) getWindowedQuery(ctx context.Context, params *events.UsageParams) (string, []interface{}) {
bucketWindow := formatWindowSizeWithBillingAnchor(params.BucketSize, params.BillingAnchor, params.Timezone)
externalCustomerFilter, customerFilter, customerArgs := buildUsageEventCustomerFilters(params)
filterConditions, filterArgs := buildFilterConditions(params.Filters)
timeConditions, timeArgs := buildTimeConditions(params)
// When GroupBy has a "properties.X" entry, return per-group rows so tiered pricing
// can be applied per group (e.g. per KRN). The events-table SQL only supports
// a single property dim here; multi-dim grouping is on the bucketed-meter path.
// 1. per_group CTE: max per group per bucket (e.g., MAX per krn per hour)
// 2. Return each group's value with group_key; total is sum of all group values for backward compat
if groupByProperty := events.FirstGroupByProperty(params.GroupBy); groupByProperty != "" && validateGroupByProperty(groupByProperty) == nil {
// groupByProperty has already been validated by validateGroupByProperty
// (alphanumeric/underscore/dot only), so it is safe to interpolate directly
// into the JSONExtractString key argument here — it cannot carry SQL metacharacters.
groupByExpr := fmt.Sprintf("JSONExtractString(assumeNotNull(properties), '%s')", groupByProperty)
query := fmt.Sprintf(`
WITH per_group AS (
SELECT
%s as bucket_start,
%s as group_key,
max(JSONExtractFloat(assumeNotNull(properties), ?)) as group_value
FROM events FINAL
PREWHERE tenant_id = ?
AND environment_id = ?
AND event_name = ?
%s
%s
%s
%s
GROUP BY bucket_start, group_key
)
SELECT
(SELECT sum(group_value) FROM per_group) as total,
bucket_start as timestamp,
group_value as value,
group_key
FROM per_group
ORDER BY bucket_start, group_key
`,
bucketWindow,
groupByExpr,
externalCustomerFilter,
customerFilter,
filterConditions,
timeConditions)
args := appendArgs([]interface{}{params.PropertyName, types.GetTenantID(ctx), types.GetEnvironmentID(ctx), params.EventName}, customerArgs, filterArgs, timeArgs)
return query, args
}
// First get max values per bucket, then sum across all buckets
query := fmt.Sprintf(`
WITH bucket_maxes AS (
SELECT
%s as bucket_start,
max(JSONExtractFloat(assumeNotNull(properties), ?)) as bucket_max
FROM events FINAL
PREWHERE tenant_id = ?
AND environment_id = ?
AND event_name = ?
%s
%s
%s
%s
GROUP BY bucket_start
ORDER BY bucket_start
)
SELECT
(SELECT sum(bucket_max) FROM bucket_maxes) as total,
bucket_start as timestamp,
bucket_max as value
FROM bucket_maxes
ORDER BY bucket_start
`,
bucketWindow,
externalCustomerFilter,
customerFilter,
filterConditions,
timeConditions)
args := appendArgs([]interface{}{params.PropertyName, types.GetTenantID(ctx), types.GetEnvironmentID(ctx), params.EventName}, customerArgs, filterArgs, timeArgs)
return query, args
}
func (a *MaxAggregator) GetType() types.AggregationType {
return types.AggregationMax
}
// WeightedSumAggregator implements weighted sum aggregation
type WeightedSumAggregator struct{}
func (a *WeightedSumAggregator) GetQuery(ctx context.Context, params *events.UsageParams) (string, []interface{}) {
windowSize := formatWindowSizeWithBillingAnchor(params.WindowSize, params.BillingAnchor, params.Timezone)
selectClause := ""
windowClause := ""
groupByClause := ""
if windowSize != "" {
selectClause = "window_size,"
windowClause = fmt.Sprintf("%s AS window_size,", windowSize)
groupByClause = "GROUP BY window_size ORDER BY window_size"
}
externalCustomerFilter, customerFilter, customerArgs := buildUsageEventCustomerFilters(params)
filterConditions, filterArgs := buildFilterConditions(params.Filters)
timeConditions, timeArgs := buildTimeConditions(params)
query := fmt.Sprintf(`
WITH
? AS period_start,
? AS period_end,
dateDiff('second', period_start, period_end) AS total_seconds
SELECT
%s sum(
(JSONExtractFloat(assumeNotNull(properties), ?) / nullIf(total_seconds, 0)) *
dateDiff('second', timestamp, period_end)
) AS total
FROM (
SELECT
%s timestamp,
properties
FROM events
PREWHERE tenant_id = ?
AND environment_id = ?
AND event_name = ?
%s
%s
%s
%s
)
%s
`,
selectClause,
windowClause,
externalCustomerFilter,
customerFilter,
filterConditions,
timeConditions,
groupByClause,
)
// Bind period_start/period_end as time.Time so the driver emits an explicit UTC
// timezone. See parseTimeConditions for the full rationale.
args := appendArgs([]interface{}{
params.StartTime.UTC(),
params.EndTime.UTC(),
params.PropertyName,
types.GetTenantID(ctx),
types.GetEnvironmentID(ctx),
params.EventName,
}, customerArgs, filterArgs, timeArgs)
return query, args
}
func (a *WeightedSumAggregator) GetType() types.AggregationType {
return types.AggregationWeightedSum
}