streaming/clickhouse/event_query_v2.go

package clickhouse

import (
	"github.com/huandu/go-sqlbuilder"
	"github.com/samber/lo"

	"github.com/openmeterio/openmeter/openmeter/streaming"
	"github.com/openmeterio/openmeter/pkg/sortx"
)

const eventQueryV2DefaultLimit = 100

// queryEventsTableV2 struct holds the parameters for v2 event queries
type queryEventsTableV2 struct {
	Database        string
	EventsTableName string
	Params          streaming.ListEventsV2Params
}

// toSQL generates the SQL query and arguments for fetching events with v2 filtering
func (q queryEventsTableV2) toSQL() (string, []interface{}) {
	tableName := getTableName(q.Database, q.EventsTableName)

	query := sqlbuilder.ClickHouse.NewSelectBuilder()
	query.Select("id", "type", "subject", "source", "time", "data", "ingested_at", "stored_at", "store_row_id")

	// Select customer_id column if customer filter is provided
	if q.Params.Customers != nil {
		query = selectCustomerIdColumn(q.EventsTableName, *q.Params.Customers, query)
	}

	query.From(tableName)

	// Base filter for namespace
	query.Where(query.Equal("namespace", q.Params.Namespace))

	if q.Params.ID != nil {
		expr := q.Params.ID.SelectWhereExpr("id", query)
		if expr != "" {
			query.Where(expr)
		}
	}

	if q.Params.Source != nil {
		expr := q.Params.Source.SelectWhereExpr("source", query)
		if expr != "" {
			query.Where(expr)
		}
	}

	if q.Params.Subject != nil {
		expr := q.Params.Subject.SelectWhereExpr("subject", query)
		if expr != "" {
			query.Where(expr)
		}
	}

	if q.Params.Customers != nil {
		query = customersWhere(tableName, *q.Params.Customers, query)
	}

	if q.Params.Type != nil {
		expr := q.Params.Type.SelectWhereExpr("type", query)
		if expr != "" {
			query.Where(expr)
		}
	}

	if q.Params.Time != nil {
		expr := q.Params.Time.SelectWhereExpr("time", query)
		if expr != "" {
			query.Where(expr)
		}
	}

	if q.Params.IngestedAt != nil {
		expr := q.Params.IngestedAt.SelectWhereExpr("ingested_at", query)
		if expr != "" {
			query.Where(expr)
		}
	}

	if q.Params.StoredAt != nil {
		expr := q.Params.StoredAt.SelectWhereExpr("stored_at", query)
		if expr != "" {
			query.Where(expr)
		}
	}

	sortCol := string(q.Params.SortBy)
	if sortCol == "" {
		sortCol = string(streaming.EventSortFieldTime)
	}

	if q.Params.Cursor != nil {
		if q.Params.SortOrder == sortx.OrderAsc {
			query.Where(
				// First filter by sort column
				query.GreaterEqualThan(sortCol, q.Params.Cursor.Time.Unix()),
				// Tie-break by store_row_id (per-row unique ULID) so same-second
				// events are not skipped across page boundaries.
				query.Or(
					query.GreaterThan(sortCol, q.Params.Cursor.Time.Unix()),
					query.GreaterThan("store_row_id", q.Params.Cursor.ID),
				),
			)
		} else {
			query.Where(
				// First filter by sort column
				query.LessEqualThan(sortCol, q.Params.Cursor.Time.Unix()),
				// Tie-break by store_row_id (per-row unique ULID) so same-second
				// events are not skipped across page boundaries.
				query.Or(
					query.LessThan(sortCol, q.Params.Cursor.Time.Unix()),
					query.LessThan("store_row_id", q.Params.Cursor.ID),
				),
			)
		}
	}

	switch q.Params.SortOrder {
	case sortx.OrderAsc:
		query.OrderByAsc(sortCol).OrderByAsc("store_row_id")
	case sortx.OrderDesc:
		fallthrough
	default:
		query.OrderByDesc(sortCol).OrderByDesc("store_row_id")
	}

	// Apply limit
	query.Limit(lo.FromPtrOr(q.Params.Limit, eventQueryV2DefaultLimit))

	return query.Build()
}

// toCountRowSQL returns the SQL query for the estimated number of rows for tracking progress
func (q queryEventsTableV2) toCountRowSQL() (string, []interface{}) {
	tableName := getTableName(q.Database, q.EventsTableName)

	query := sqlbuilder.ClickHouse.NewSelectBuilder()
	query.Select("count() as total")
	query.From(tableName)

	// Base filter for namespace
	query.Where(query.Equal("namespace", q.Params.Namespace))

	// Apply basic filters that might significantly affect the count
	// Note: We don't include all filters here for performance reasons

	if q.Params.Type != nil {
		expr := q.Params.Type.SelectWhereExpr("type", query)
		if expr != "" {
			query.Where(expr)
		}
	}

	if q.Params.Subject != nil {
		expr := q.Params.Subject.SelectWhereExpr("subject", query)
		if expr != "" {
			query.Where(expr)
		}
	}

	if q.Params.Customers != nil {
		query = customersWhere(tableName, *q.Params.Customers, query)
	}

	if q.Params.Time != nil {
		expr := q.Params.Time.SelectWhereExpr("time", query)
		if expr != "" {
			query.Where(expr)
		}
	}

	return query.Build()
}