// Copyright 2026 Arctel.net // SPDX-License-Identifier: Apache-2.0 package analytics import ( "context" "fmt" "strings" "time" analyticsmodel "github.com/Rain-kl/Wavelet/internal/model/analytics" ) // NodeAccessLogBucketAggregate is a folded bucket aggregate row. type NodeAccessLogBucketAggregate = analyticsmodel.NodeAccessLogBucketAggregate // NodeAccessLogWAFIPAggregate is a per-IP aggregate row for WAF automatic rules. type NodeAccessLogWAFIPAggregate = analyticsmodel.NodeAccessLogWAFIPAggregate // NodeAccessLogBucketDimension is a bucket dimension value. type NodeAccessLogBucketDimension = analyticsmodel.NodeAccessLogBucketDimension // NodeAccessLogIPAggregate is an IP aggregate row. type NodeAccessLogIPAggregate = analyticsmodel.NodeAccessLogIPAggregate // NodeAccessLogIPSummary is an IP summary row. type NodeAccessLogIPSummary = analyticsmodel.NodeAccessLogIPSummary // NodeAccessLogIPTrend is an IP trend bucket row. type NodeAccessLogIPTrend = analyticsmodel.NodeAccessLogIPTrend // BucketAggregatesNodeAccessLogs returns folded bucket aggregates with unique IP/host counts. func BucketAggregatesNodeAccessLogs(ctx context.Context, filter NodeAccessLogFilter, bucketSeconds int64) ([]NodeAccessLogBucketAggregate, error) { conn, err := nodeAccessLogConn() if err != nil { return nil, err } clause, args := buildNodeAccessLogFilterClause(filter) bucketExpr := nodeAccessLogBucketEpochExpr(bucketSeconds) tableName := nodeAccessLogTableName() sql := fmt.Sprintf(` SELECT %s AS bucket_epoch, count() AS request_count, countIf(status_code < 400) AS success_count, countIf(status_code >= 400 AND status_code < 500) AS client_error_count, countIf(status_code >= 500) AS server_error_count, countIf(status_code >= 200 AND status_code < 300) AS status_2xx_count, countIf(status_code >= 400 AND status_code < 500) AS status_4xx_count, countIf(status_code >= 500) AS status_5xx_count, uniqExactIf(remote_addr, remote_addr != '') AS unique_ip_count, uniqExactIf(host, host != '') AS unique_host_count, sum(bytes_sent) AS bytes_sent, sum(request_length) AS request_length FROM %s WHERE %s GROUP BY bucket_epoch ORDER BY %s`, bucketExpr, tableName, clause, nodeAccessLogBucketOrderClause(filter.SortBy, filter.SortOrder)) if filter.PageSize > 0 { if filter.Page < 0 { filter.Page = 0 } sql += clickHouseLimitOffsetClause args = append(args, filter.PageSize, filter.Page*filter.PageSize) } rows, err := conn.Query(ctx, sql, args...) if err != nil { return nil, fmt.Errorf("bucket aggregates node access logs: %w", err) } defer func() { _ = rows.Close() }() var result []NodeAccessLogBucketAggregate for rows.Next() { var ( bucketEpoch int64 requestCount, successCount, clientErrorCount, serverErrorCount, status2xxCount, status4xxCount, status5xxCount, uniqueIPCount, uniqueHostCount, bytesSent, requestLength uint64 ) if err := rows.Scan(&bucketEpoch, &requestCount, &successCount, &clientErrorCount, &serverErrorCount, &status2xxCount, &status4xxCount, &status5xxCount, &uniqueIPCount, &uniqueHostCount, &bytesSent, &requestLength); err != nil { return nil, fmt.Errorf("scan bucket aggregate row: %w", err) } result = append(result, NodeAccessLogBucketAggregate{ BucketEpoch: bucketEpoch, RequestCount: safeInt64Count(requestCount), SuccessCount: safeInt64Count(successCount), ClientErrorCount: safeInt64Count(clientErrorCount), ServerErrorCount: safeInt64Count(serverErrorCount), Status2xxCount: safeInt64Count(status2xxCount), Status4xxCount: safeInt64Count(status4xxCount), Status5xxCount: safeInt64Count(status5xxCount), UniqueIPCount: safeInt64Count(uniqueIPCount), UniqueHostCount: safeInt64Count(uniqueHostCount), BytesSent: safeInt64Count(bytesSent), RequestLength: safeInt64Count(requestLength), }) } return result, nil } // CountBucketAggregatesNodeAccessLogs returns the number of folded buckets matching filter. func CountBucketAggregatesNodeAccessLogs(ctx context.Context, filter NodeAccessLogFilter, bucketSeconds int64) (int64, error) { conn, err := nodeAccessLogConn() if err != nil { return 0, err } clause, args := buildNodeAccessLogFilterClause(filter) bucketExpr := nodeAccessLogBucketEpochExpr(bucketSeconds) tableName := nodeAccessLogTableName() sql := fmt.Sprintf(` SELECT count() FROM ( SELECT 1 FROM %s WHERE %s GROUP BY %s )`, tableName, clause, bucketExpr) var totalBuckets uint64 if err := conn.QueryRow(ctx, sql, args...).Scan(&totalBuckets); err != nil { return 0, fmt.Errorf("count bucket aggregates node access logs: %w", err) } return safeInt64Count(totalBuckets), nil } // BucketDimensionsNodeAccessLogs returns bucket dimension values. func BucketDimensionsNodeAccessLogs(ctx context.Context, filter NodeAccessLogFilter, column string, bucketSeconds int64) ([]NodeAccessLogBucketDimension, error) { conn, err := nodeAccessLogConn() if err != nil { return nil, err } clause, args := buildNodeAccessLogFilterClause(filter) bucketExpr := nodeAccessLogBucketEpochExpr(bucketSeconds) tableName := nodeAccessLogTableName() sql := fmt.Sprintf(` SELECT %s AS bucket_epoch, trim(%s) AS value FROM %s WHERE %s AND trim(%s) != '' GROUP BY bucket_epoch, trim(%s)`, bucketExpr, column, tableName, clause, column, column) rows, err := conn.Query(ctx, sql, args...) if err != nil { return nil, fmt.Errorf("bucket dimensions node access logs: %w", err) } defer func() { _ = rows.Close() }() var result []NodeAccessLogBucketDimension for rows.Next() { var item NodeAccessLogBucketDimension if err := rows.Scan(&item.BucketEpoch, &item.Value); err != nil { return nil, fmt.Errorf("scan bucket dimension row: %w", err) } result = append(result, item) } return result, nil } // IPAggregatesNodeAccessLogs returns IP aggregate rows. func IPAggregatesNodeAccessLogs(ctx context.Context, filter NodeAccessLogFilter, exactRemoteAddr bool) ([]NodeAccessLogIPAggregate, error) { conn, err := nodeAccessLogConn() if err != nil { return nil, err } clause, args := buildNodeAccessLogFilterClause(filter) queryClause := clause queryArgs := append([]any{}, args...) if exactRemoteAddr { trimmed := normalizeNodeAccessLogRemoteAddr(filter.RemoteAddr) if trimmed == "" { return []NodeAccessLogIPAggregate{}, nil } queryClause = combineNodeAccessLogSQLClauses(queryClause, "remote_addr = ?") queryArgs = append(queryArgs, trimmed) } lastSeenExpr := nodeAccessLogEpochExpr() tableName := nodeAccessLogTableName() sql := fmt.Sprintf(` SELECT remote_addr, count() AS request_count, countIf(status_code < 400) AS success_count, countIf(status_code >= 400 AND status_code < 500) AS client_error_count, countIf(status_code >= 500) AS server_error_count, max(%s) AS last_seen_epoch FROM %s WHERE %s AND remote_addr != '' GROUP BY remote_addr`, lastSeenExpr, tableName, queryClause) rows, err := conn.Query(ctx, sql, queryArgs...) if err != nil { return nil, fmt.Errorf("ip aggregates node access logs: %w", err) } defer func() { _ = rows.Close() }() var result []NodeAccessLogIPAggregate for rows.Next() { var ( remoteAddr string lastSeenEpoch int64 requestCount, successCount, clientErrorCount, serverErrorCount uint64 ) if err := rows.Scan(&remoteAddr, &requestCount, &successCount, &clientErrorCount, &serverErrorCount, &lastSeenEpoch); err != nil { return nil, fmt.Errorf("scan ip aggregate row: %w", err) } result = append(result, NodeAccessLogIPAggregate{ RemoteAddr: remoteAddr, RequestCount: safeInt64Count(requestCount), SuccessCount: safeInt64Count(successCount), ClientErrorCount: safeInt64Count(clientErrorCount), ServerErrorCount: safeInt64Count(serverErrorCount), LastSeenEpoch: lastSeenEpoch, }) } return result, nil } // IPSummariesNodeAccessLogs returns paginated IP summary rows for the filter window. // recentSince is ignored (kept for call-site compatibility); recent_requests is always 0. func IPSummariesNodeAccessLogs(ctx context.Context, filter NodeAccessLogFilter, _ time.Time) ([]NodeAccessLogIPSummary, error) { conn, err := nodeAccessLogConn() if err != nil { return nil, err } clause, args := buildNodeAccessLogFilterClause(filter) lastSeenExpr := nodeAccessLogEpochExpr() queryArgs := append([]any{}, args...) tableName := nodeAccessLogTableName() // Outer query allows ORDER BY success_ratio without repeating countIf. innerSQL := fmt.Sprintf(` SELECT remote_addr, argMax(region, logged_at) AS region, count() AS total_requests, countIf(status_code >= 200 AND status_code < 300) AS success_2xx_count, sum(request_length) AS request_length, sum(bytes_sent) AS bytes_sent, max(%s) AS last_seen_epoch FROM %s WHERE %s AND remote_addr != '' GROUP BY remote_addr`, lastSeenExpr, tableName, clause) sql := fmt.Sprintf(` SELECT remote_addr, region, total_requests, success_2xx_count, if(total_requests = 0, 0., toFloat64(success_2xx_count) / toFloat64(total_requests)) AS success_ratio, request_length, bytes_sent, last_seen_epoch FROM (%s) ORDER BY %s`, innerSQL, nodeAccessLogIPSummaryOrderClause(filter.SortBy, filter.SortOrder)) if filter.PageSize > 0 { if filter.Page < 0 { filter.Page = 0 } sql += clickHouseLimitOffsetClause queryArgs = append(queryArgs, filter.PageSize, filter.Page*filter.PageSize) } rows, err := conn.Query(ctx, sql, queryArgs...) if err != nil { return nil, fmt.Errorf("ip summaries node access logs: %w", err) } defer func() { _ = rows.Close() }() var result []NodeAccessLogIPSummary for rows.Next() { var ( remoteAddr, region string lastSeenEpoch int64 successRatio float64 totalRequests, success2xx, bytesReceived, bytes uint64 ) if err := rows.Scan( &remoteAddr, ®ion, &totalRequests, &success2xx, &successRatio, &bytesReceived, &bytes, &lastSeenEpoch, ); err != nil { return nil, fmt.Errorf("scan ip summary row: %w", err) } result = append(result, NodeAccessLogIPSummary{ RemoteAddr: remoteAddr, Region: region, TotalRequests: safeInt64Count(totalRequests), Success2xxCount: safeInt64Count(success2xx), SuccessRatio: successRatio, BytesReceived: safeInt64Count(bytesReceived), BytesSent: safeInt64Count(bytes), RecentRequests: 0, LastSeenEpoch: lastSeenEpoch, }) } return result, nil } // CountIPSummaryNodeAccessLogs returns the number of distinct IPs matching filter. func CountIPSummaryNodeAccessLogs(ctx context.Context, filter NodeAccessLogFilter) (int64, error) { conn, err := nodeAccessLogConn() if err != nil { return 0, err } clause, args := buildNodeAccessLogFilterClause(filter) tableName := nodeAccessLogTableName() sql := fmt.Sprintf(` SELECT count() FROM ( SELECT 1 FROM %s WHERE %s AND remote_addr != '' GROUP BY remote_addr )`, tableName, clause) var totalIPs uint64 if err := conn.QueryRow(ctx, sql, args...).Scan(&totalIPs); err != nil { return 0, fmt.Errorf("count ip summary node access logs: %w", err) } return safeInt64Count(totalIPs), nil } // IPAggregatesForWAFNodeAccessLogs returns per-IP aggregates for WAF automatic rules. func IPAggregatesForWAFNodeAccessLogs(ctx context.Context, filter NodeAccessLogFilter) ([]NodeAccessLogWAFIPAggregate, error) { conn, err := nodeAccessLogConn() if err != nil { return nil, err } clause, args := buildNodeAccessLogFilterClause(filter) lastSeenExpr := nodeAccessLogEpochExpr() hostIsIPExpr := nodeAccessLogHostIsIPLiteralExpr() tableName := nodeAccessLogTableName() sql := fmt.Sprintf(` SELECT remote_addr, count() AS request_count, countIf(status_code = 404) AS status_404_count, countIf(status_code >= 400 AND status_code < 500) AS client_error_count, countIf(status_code >= 500) AS server_error_count, countIf(%s) AS ip_host_count, max(%s) AS last_seen_epoch FROM %s WHERE %s AND remote_addr != '' GROUP BY remote_addr`, hostIsIPExpr, lastSeenExpr, tableName, clause) rows, err := conn.Query(ctx, sql, args...) if err != nil { return nil, fmt.Errorf("ip aggregates for waf node access logs: %w", err) } defer func() { _ = rows.Close() }() aggregates := make(map[string]*NodeAccessLogWAFIPAggregate) order := make([]string, 0) for rows.Next() { var ( remoteAddr string lastSeenEpoch int64 requestCount, status404Count, clientErrorCount, serverErrorCount, ipHostCount uint64 ) if err := rows.Scan(&remoteAddr, &requestCount, &status404Count, &clientErrorCount, &serverErrorCount, &ipHostCount, &lastSeenEpoch); err != nil { return nil, fmt.Errorf("scan waf ip aggregate row: %w", err) } remoteAddr = strings.TrimSpace(remoteAddr) if remoteAddr == "" { continue } aggregates[remoteAddr] = &NodeAccessLogWAFIPAggregate{ RemoteAddr: remoteAddr, RequestCount: safeInt64Count(requestCount), Status404Count: safeInt64Count(status404Count), ClientErrorCount: safeInt64Count(clientErrorCount), ServerErrorCount: safeInt64Count(serverErrorCount), IPHostCount: safeInt64Count(ipHostCount), LastSeenEpoch: lastSeenEpoch, StatusCounts: make(map[int]int64), } order = append(order, remoteAddr) } if err := mergeWAFIPStatusCodeCounts(ctx, filter, aggregates); err != nil { return nil, err } result := make([]NodeAccessLogWAFIPAggregate, 0, len(order)) for _, remoteAddr := range order { if aggregate := aggregates[remoteAddr]; aggregate != nil { result = append(result, *aggregate) } } return result, nil } func mergeWAFIPStatusCodeCounts(ctx context.Context, filter NodeAccessLogFilter, aggregates map[string]*NodeAccessLogWAFIPAggregate) error { if len(aggregates) == 0 { return nil } conn, err := nodeAccessLogConn() if err != nil { return err } clause, args := buildNodeAccessLogFilterClause(filter) tableName := nodeAccessLogTableName() sql := fmt.Sprintf(` SELECT remote_addr, status_code, count() AS status_count FROM %s WHERE %s AND remote_addr != '' GROUP BY remote_addr, status_code`, tableName, clause) rows, err := conn.Query(ctx, sql, args...) if err != nil { return fmt.Errorf("waf ip status code counts: %w", err) } defer func() { _ = rows.Close() }() for rows.Next() { var ( remoteAddr string statusCode int32 statusCount uint64 ) if err := rows.Scan(&remoteAddr, &statusCode, &statusCount); err != nil { return fmt.Errorf("scan waf ip status code row: %w", err) } remoteAddr = strings.TrimSpace(remoteAddr) aggregate := aggregates[remoteAddr] if aggregate == nil { continue } if aggregate.StatusCounts == nil { aggregate.StatusCounts = make(map[int]int64) } aggregate.StatusCounts[int(statusCode)] = safeInt64Count(statusCount) } return nil } // IPTrendNodeAccessLogs returns IP trend bucket rows. func IPTrendNodeAccessLogs(ctx context.Context, filter NodeAccessLogFilter, bucketSeconds int64) ([]NodeAccessLogIPTrend, error) { conn, err := nodeAccessLogConn() if err != nil { return nil, err } clause, args := buildNodeAccessLogFilterClause(filter) bucketExpr := nodeAccessLogBucketEpochExpr(bucketSeconds) tableName := nodeAccessLogTableName() sql := fmt.Sprintf(` SELECT %s AS bucket_epoch, count() AS request_count FROM %s WHERE %s GROUP BY bucket_epoch ORDER BY bucket_epoch ASC`, bucketExpr, tableName, clause) rows, err := conn.Query(ctx, sql, args...) if err != nil { return nil, fmt.Errorf("ip trend node access logs: %w", err) } defer func() { _ = rows.Close() }() var result []NodeAccessLogIPTrend for rows.Next() { var ( bucketEpoch int64 requestCount uint64 ) if err := rows.Scan(&bucketEpoch, &requestCount); err != nil { return nil, fmt.Errorf("scan ip trend row: %w", err) } result = append(result, NodeAccessLogIPTrend{ BucketEpoch: bucketEpoch, RequestCount: safeInt64Count(requestCount), }) } return result, nil }