internal/repository/clickhouse/reports.go
Rama main · réplica de lectura del repositorio original.
package clickhouse
import (
"cmp"
"context"
"crypto/sha256"
"encoding/binary"
"encoding/hex"
"encoding/json"
"errors"
"hash"
"maps"
"slices"
"strings"
"time"
clickhouse_go "github.com/ClickHouse/clickhouse-go/v2"
"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
"github.com/ClickHouse/clickhouse-go/v2/lib/proto"
"github.com/flexprice/flexprice/internal/cache"
"github.com/flexprice/flexprice/internal/config"
"github.com/flexprice/flexprice/internal/domain/analytics"
ierr "github.com/flexprice/flexprice/internal/errors"
"github.com/flexprice/flexprice/internal/logger"
"github.com/flexprice/flexprice/internal/types"
)
const (
defaultReportsMaxExecutionSeconds = 10
defaultReportsMaxResultRows = 10000
maxCachedReportBytes = 1 << 20
)
var _ analytics.ReportsRepository = (*ReportsRepository)(nil)
type ReportsRepository struct {
conn driver.Conn
cfg config.ReportsConfig
logger *logger.Logger
cache cache.RedisCache
source string
usageConn driver.Conn
usageSource string
}
func NewReportsRepository(cfg *config.Configuration, log *logger.Logger, rc cache.RedisCache) (*ReportsRepository, error) {
r := &ReportsRepository{cfg: cfg.Reports, logger: log, cache: rc}
if r.cfg.MaxExecutionSeconds <= 0 {
r.cfg.MaxExecutionSeconds = defaultReportsMaxExecutionSeconds
}
if r.cfg.MaxResultRows <= 0 {
r.cfg.MaxResultRows = defaultReportsMaxResultRows
}
if !r.cfg.Enabled() {
return r, nil
}
opts := cfg.ClickHouse.GetClientOptions()
if r.cfg.Address != "" {
opts.Addr = []string{r.cfg.Address}
}
r.source = strings.Join(opts.Addr, ",")
opts.Auth = clickhouse_go.Auth{
Database: r.cfg.Database,
Username: r.cfg.Username,
Password: r.cfg.Password,
}
conn, err := clickhouse_go.Open(opts)
if err != nil {
return nil, ierr.WithError(err).WithHint("open reports clickhouse connection").Mark(ierr.ErrSystem)
}
r.conn = conn
if !r.cfg.UsageOnEventsClickHouse {
return r, nil
}
r.cfg.UsageDatabase = cmp.Or(r.cfg.UsageDatabase, cfg.ClickHouse.Database)
usageOpts := cfg.ClickHouse.GetClientOptions()
usageOpts.Auth.Database = r.cfg.UsageDatabase
usageConn, err := clickhouse_go.Open(usageOpts)
if err != nil {
_ = conn.Close()
return nil, ierr.WithError(err).WithHint("open usage clickhouse connection").Mark(ierr.ErrSystem)
}
r.usageConn = usageConn
r.usageSource = strings.Join(usageOpts.Addr, ",")
return r, nil
}
func (r *ReportsRepository) Enabled() bool { return r.conn != nil }
func (r *ReportsRepository) MaxRows() int { return r.cfg.MaxResultRows }
const (
chCodeTooManyRows = 158
chCodeTimeoutExceeded = 159
chCodeMemoryLimitExceeded = 241
)
func reportError(err error, hint string) error {
var ex *proto.Exception
if errors.As(err, &ex) && (ex.Code == chCodeTooManyRows || ex.Code == chCodeTimeoutExceeded || ex.Code == chCodeMemoryLimitExceeded) {
return ierr.WithError(err).
WithHint("Query too large; narrow the date range").
Mark(ierr.ErrValidation)
}
return ierr.WithError(err).WithHint(hint).Mark(ierr.ErrDatabase)
}
func (r *ReportsRepository) RunReport(ctx context.Context, rv *analytics.ResolvedView, tenantID, environmentID string) (*analytics.ReportResult, error) {
if r.conn == nil {
return nil, ierr.NewError("financial reports are not enabled").Mark(ierr.ErrPermissionDenied)
}
q, err := r.buildReport(rv, tenantID, environmentID)
if err != nil {
return nil, err
}
rows, err := r.forReport(rv).run(ctx, q)
if err != nil {
return nil, err
}
return reportResult(q, rows, r.cfg.MaxResultRows)
}
// forReport targets the usage server for usage-backed metrics.
func (r *ReportsRepository) forReport(rv *analytics.ResolvedView) *ReportsRepository {
if r.usageConn == nil || rv == nil || !slices.Contains(rv.Metrics, types.MetricEventsVolume) {
return r
}
u := *r
u.conn = r.usageConn
u.source = r.usageSource
return &u
}
func (r *ReportsRepository) buildReport(rv *analytics.ResolvedView, tenantID, environmentID string) (*reportQuery, error) {
if rv == nil {
return buildBillingQueryWithUsage(rv, r.cfg.Database, r.cfg.UsageDatabase, tenantID, environmentID, r.cfg.MaxResultRows)
}
w := *rv
w.Time.From = r.stableTime(rv.Time.From)
w.Time.To = r.stableTime(rv.Time.To)
return buildBillingQueryWithUsage(&w, r.cfg.Database, r.cfg.UsageDatabase, tenantID, environmentID, r.cfg.MaxResultRows)
}
// stableTime lets relative ranges share cache keys.
func (r *ReportsRepository) stableTime(t time.Time) time.Time {
if !r.resultCacheOn() {
return t
}
return t.Truncate(time.Minute)
}
func (r *ReportsRepository) resultCacheOn() bool {
return r.cache != nil && r.cache.IsEnabled() && r.cfg.QueryCacheTTLSeconds > 0
}
func reportResult(q *reportQuery, rows [][]string, maxRows int) (*analytics.ReportResult, error) {
if !q.TopN && maxRows > 0 && len(rows) > maxRows {
return nil, ierr.NewErrorf("result exceeds %d rows", maxRows).
WithHint("Narrow the time range or remove dimensions").
Mark(ierr.ErrValidation)
}
return &analytics.ReportResult{Columns: q.Columns, Rows: rows}, nil
}
func (r *ReportsRepository) run(ctx context.Context, q *reportQuery) ([][]string, error) {
return r.cached(ctx, q, func() ([][]string, error) { return r.query(ctx, q) })
}
func (r *ReportsRepository) cached(ctx context.Context, q *reportQuery, fetch func() ([][]string, error)) ([][]string, error) {
if !r.resultCacheOn() {
return fetch()
}
key := reportCacheKey(ctx, r.source, q)
if v, ok := r.cache.Get(ctx, key); ok {
if rows, ok := cache.UnmarshalCacheValue[[][]string](v); ok {
return *rows, nil
}
}
rows, err := fetch()
if err != nil {
return nil, err
}
if b, err := json.Marshal(rows); err == nil && len(b) <= maxCachedReportBytes {
r.cache.Set(ctx, key, string(b), time.Duration(r.cfg.QueryCacheTTLSeconds)*time.Second)
}
return rows, nil
}
// Params carry tenant and environment.
func reportCacheKey(ctx context.Context, source string, q *reportQuery) string {
h := sha256.New()
writeLengthPrefixed(h, source)
writeLengthPrefixed(h, q.SQL)
for _, k := range slices.Sorted(maps.Keys(q.Params)) {
writeLengthPrefixed(h, k)
writeLengthPrefixed(h, q.Params[k])
}
return cache.GenerateKey(ctx, cache.PrefixReportResult, hex.EncodeToString(h.Sum(nil)))
}
func writeLengthPrefixed(h hash.Hash, s string) {
var n [8]byte
binary.BigEndian.PutUint64(n[:], uint64(len(s)))
h.Write(n[:])
h.Write([]byte(s))
}
func (r *ReportsRepository) query(ctx context.Context, q *reportQuery) ([][]string, error) {
settings := clickhouse_go.Settings{
"max_execution_time": r.cfg.MaxExecutionSeconds,
"max_result_rows": r.cfg.MaxResultRows + 1,
"result_overflow_mode": "throw",
// Blocks writes; still allows settings.
"readonly": 2,
}
if r.cfg.MaxRowsToRead > 0 {
settings["max_rows_to_read"] = r.cfg.MaxRowsToRead
}
if r.cfg.QueryCacheTTLSeconds > 0 {
settings["use_query_cache"] = 1
settings["query_cache_ttl"] = r.cfg.QueryCacheTTLSeconds
// Cache, not throw, on now().
settings["query_cache_nondeterministic_function_handling"] = "save"
} else {
settings["use_query_cache"] = 0
}
ctx = clickhouse_go.Context(ctx, clickhouse_go.WithParameters(q.Params), clickhouse_go.WithSettings(settings))
rows, err := r.conn.Query(ctx, q.SQL)
if err != nil {
return nil, reportError(err, "run report query")
}
defer rows.Close()
out := [][]string{}
for rows.Next() {
vals := make([]string, len(q.Columns))
ptrs := make([]any, len(vals))
for i := range vals {
ptrs[i] = &vals[i]
}
if err := rows.Scan(ptrs...); err != nil {
return nil, ierr.WithError(err).WithHint("scan report row").Mark(ierr.ErrDatabase)
}
out = append(out, vals)
}
if err := rows.Err(); err != nil {
return nil, reportError(err, "read report rows")
}
return out, nil
}