OpenFlare/internal/repository/analytics/node_observability.go

// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0

package analytics

import (
	"context"
	"fmt"

	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
	"github.com/Rain-kl/Wavelet/internal/db"
	analyticsmodel "github.com/Rain-kl/Wavelet/internal/model/analytics"
)

func observabilityConn() (driver.Conn, error) {
	if db.ChConn == nil {
		return nil, fmt.Errorf("clickhouse connection is not initialized")
	}
	return db.ChConn, nil
}

// ListNodeMetricSnapshots returns metric snapshots matching filter.
func ListNodeMetricSnapshots(ctx context.Context, filter NodeObservabilityFilter) ([]analyticsmodel.NodeMetricSnapshot, error) {
	conn, err := observabilityConn()
	if err != nil {
		return nil, err
	}
	clause, args := buildNodeObservabilityFilterClause(filter, "captured_at")
	tableName := nodeMetricSnapshotTableName()
	sql := fmt.Sprintf(`
SELECT id, node_id, captured_at, cpu_usage_percent, memory_used_bytes, memory_total_bytes, storage_used_bytes, storage_total_bytes, disk_read_bytes, disk_write_bytes, network_rx_bytes, network_tx_bytes, created_at
FROM %s
WHERE %s
ORDER BY %s`, tableName, clause, nodeObservabilityCapturedAtOrderClause())
	if filter.Limit > 0 {
		sql += clickHouseLimitClause
		args = append(args, filter.Limit)
	}
	rows, err := conn.Query(ctx, sql, args...)
	if err != nil {
		return nil, fmt.Errorf("list node metric snapshots: %w", err)
	}
	defer func() { _ = rows.Close() }()
	return scanNodeMetricSnapshotRows(rows)
}

// ListNodeRequestReports returns request reports matching filter.
func ListNodeRequestReports(ctx context.Context, filter NodeObservabilityFilter) ([]analyticsmodel.NodeRequestReport, error) {
	conn, err := observabilityConn()
	if err != nil {
		return nil, err
	}
	clause, args := buildNodeObservabilityFilterClause(filter, "window_ended_at")
	tableName := nodeRequestReportTableName()
	sql := fmt.Sprintf(`
SELECT id, node_id, window_started_at, window_ended_at, request_count, error_count, unique_visitor_count, status_codes_json, top_domains_json, source_countries_json, created_at
FROM %s
WHERE %s
ORDER BY %s`, tableName, clause, nodeObservabilityWindowEndedAtOrderClause())
	if filter.Limit > 0 {
		sql += clickHouseLimitClause
		args = append(args, filter.Limit)
	}
	rows, err := conn.Query(ctx, sql, args...)
	if err != nil {
		return nil, fmt.Errorf("list node request reports: %w", err)
	}
	defer func() { _ = rows.Close() }()
	return scanNodeRequestReportRows(rows)
}

// ListNodeObsOpenresty returns OpenResty observations matching filter.
func ListNodeObsOpenresty(ctx context.Context, filter NodeObservabilityFilter) ([]analyticsmodel.NodeObsOpenresty, error) {
	conn, err := observabilityConn()
	if err != nil {
		return nil, err
	}
	clause, args := buildNodeObservabilityFilterClause(filter, "captured_at")
	tableName := nodeObsOpenrestyTableName()
	sql := fmt.Sprintf(`
SELECT id, node_id, captured_at, openresty_rx_bytes, openresty_tx_bytes, openresty_connections, created_at
FROM %s
WHERE %s
ORDER BY %s`, tableName, clause, nodeObservabilityCapturedAtOrderClause())
	if filter.Limit > 0 {
		sql += clickHouseLimitClause
		args = append(args, filter.Limit)
	}
	rows, err := conn.Query(ctx, sql, args...)
	if err != nil {
		return nil, fmt.Errorf("list node openresty observations: %w", err)
	}
	defer func() { _ = rows.Close() }()
	return scanNodeObsOpenrestyRows(rows)
}

// ListNodeObsFrps returns FRPS observations matching filter.
func ListNodeObsFrps(ctx context.Context, filter NodeObservabilityFilter) ([]analyticsmodel.NodeObsFrps, error) {
	conn, err := observabilityConn()
	if err != nil {
		return nil, err
	}
	clause, args := buildNodeObservabilityFilterClause(filter, "captured_at")
	tableName := nodeObsFrpsTableName()
	sql := fmt.Sprintf(`
SELECT id, node_id, captured_at, frps_connections, frps_proxy_count, frps_client_count, frps_proxies, created_at
FROM %s
WHERE %s
ORDER BY %s`, tableName, clause, nodeObservabilityCapturedAtOrderClause())
	if filter.Limit > 0 {
		sql += clickHouseLimitClause
		args = append(args, filter.Limit)
	}
	rows, err := conn.Query(ctx, sql, args...)
	if err != nil {
		return nil, fmt.Errorf("list node frps observations: %w", err)
	}
	defer func() { _ = rows.Close() }()
	return scanNodeObsFrpsRows(rows)
}

// ListNodeObsFrpc returns FRPC observations matching filter.
func ListNodeObsFrpc(ctx context.Context, filter NodeObservabilityFilter) ([]analyticsmodel.NodeObsFrpc, error) {
	conn, err := observabilityConn()
	if err != nil {
		return nil, err
	}
	clause, args := buildNodeObservabilityFilterClause(filter, "captured_at")
	tableName := nodeObsFrpcTableName()
	sql := fmt.Sprintf(`
SELECT id, node_id, captured_at, tunnel_status, connected_relays_count, created_at
FROM %s
WHERE %s
ORDER BY %s`, tableName, clause, nodeObservabilityCapturedAtOrderClause())
	if filter.Limit > 0 {
		sql += clickHouseLimitClause
		args = append(args, filter.Limit)
	}
	rows, err := conn.Query(ctx, sql, args...)
	if err != nil {
		return nil, fmt.Errorf("list node frpc observations: %w", err)
	}
	defer func() { _ = rows.Close() }()
	return scanNodeObsFrpcRows(rows)
}

func scanNodeMetricSnapshotRows(rows driver.Rows) ([]analyticsmodel.NodeMetricSnapshot, error) {
	var result []analyticsmodel.NodeMetricSnapshot
	for rows.Next() {
		var item analyticsmodel.NodeMetricSnapshot
		if err := rows.Scan(
			&item.ID,
			&item.NodeID,
			&item.CapturedAt,
			&item.CPUUsagePercent,
			&item.MemoryUsedBytes,
			&item.MemoryTotalBytes,
			&item.StorageUsedBytes,
			&item.StorageTotalBytes,
			&item.DiskReadBytes,
			&item.DiskWriteBytes,
			&item.NetworkRxBytes,
			&item.NetworkTxBytes,
			&item.CreatedAt,
		); err != nil {
			return nil, fmt.Errorf("scan node metric snapshot row: %w", err)
		}
		item.CapturedAt = item.CapturedAt.UTC()
		item.CreatedAt = item.CreatedAt.UTC()
		result = append(result, item)
	}
	return result, nil
}

func scanNodeRequestReportRows(rows driver.Rows) ([]analyticsmodel.NodeRequestReport, error) {
	var result []analyticsmodel.NodeRequestReport
	for rows.Next() {
		var item analyticsmodel.NodeRequestReport
		if err := rows.Scan(
			&item.ID,
			&item.NodeID,
			&item.WindowStartedAt,
			&item.WindowEndedAt,
			&item.RequestCount,
			&item.ErrorCount,
			&item.UniqueVisitorCount,
			&item.StatusCodesJSON,
			&item.TopDomainsJSON,
			&item.SourceCountriesJSON,
			&item.CreatedAt,
		); err != nil {
			return nil, fmt.Errorf("scan node request report row: %w", err)
		}
		item.WindowStartedAt = item.WindowStartedAt.UTC()
		item.WindowEndedAt = item.WindowEndedAt.UTC()
		item.CreatedAt = item.CreatedAt.UTC()
		result = append(result, item)
	}
	return result, nil
}

func scanNodeObsOpenrestyRows(rows driver.Rows) ([]analyticsmodel.NodeObsOpenresty, error) {
	var result []analyticsmodel.NodeObsOpenresty
	for rows.Next() {
		var item analyticsmodel.NodeObsOpenresty
		if err := rows.Scan(
			&item.ID,
			&item.NodeID,
			&item.CapturedAt,
			&item.OpenrestyRxBytes,
			&item.OpenrestyTxBytes,
			&item.OpenrestyConnections,
			&item.CreatedAt,
		); err != nil {
			return nil, fmt.Errorf("scan node openresty observation row: %w", err)
		}
		item.CapturedAt = item.CapturedAt.UTC()
		item.CreatedAt = item.CreatedAt.UTC()
		result = append(result, item)
	}
	return result, nil
}

func scanNodeObsFrpsRows(rows driver.Rows) ([]analyticsmodel.NodeObsFrps, error) {
	var result []analyticsmodel.NodeObsFrps
	for rows.Next() {
		var item analyticsmodel.NodeObsFrps
		if err := rows.Scan(
			&item.ID,
			&item.NodeID,
			&item.CapturedAt,
			&item.FrpsConnections,
			&item.FrpsProxyCount,
			&item.FrpsClientCount,
			&item.FrpsProxies,
			&item.CreatedAt,
		); err != nil {
			return nil, fmt.Errorf("scan node frps observation row: %w", err)
		}
		item.CapturedAt = item.CapturedAt.UTC()
		item.CreatedAt = item.CreatedAt.UTC()
		result = append(result, item)
	}
	return result, nil
}

func scanNodeObsFrpcRows(rows driver.Rows) ([]analyticsmodel.NodeObsFrpc, error) {
	var result []analyticsmodel.NodeObsFrpc
	for rows.Next() {
		var item analyticsmodel.NodeObsFrpc
		if err := rows.Scan(
			&item.ID,
			&item.NodeID,
			&item.CapturedAt,
			&item.TunnelStatus,
			&item.ConnectedRelaysCount,
			&item.CreatedAt,
		); err != nil {
			return nil, fmt.Errorf("scan node frpc observation row: %w", err)
		}
		item.CapturedAt = item.CapturedAt.UTC()
		item.CreatedAt = item.CreatedAt.UTC()
		result = append(result, item)
	}
	return result, nil
}