HY2XS_flamy/apps/dao/dashboard.go

package dao

import (
	"errors"
	"github.com/sirupsen/logrus"
	"hy2xs-admin/model/constant"
	"hy2xs-admin/model/entity"
	"hy2xs-admin/model/vo"
	"strings"
	"time"
)

func SaveMetricSample(sample entity.MetricSample) error {
	if tx := sqliteDB.Save(&sample); tx.Error != nil {
		logrus.Errorf("%v", tx.Error)
		return errors.New(constant.SysError)
	}
	return nil
}

func LastMetricSample() (entity.MetricSample, error) {
	var sample entity.MetricSample
	if tx := sqliteDB.Model(&entity.MetricSample{}).Order("sampled_at desc").Limit(1).Find(&sample); tx.Error != nil {
		logrus.Errorf("%v", tx.Error)
		return sample, errors.New(constant.SysError)
	}
	return sample, nil
}

func CleanupMetricSample(olderThanMs int64) error {
	if !tableExists("metric_sample") {
		return nil
	}
	if tx := sqliteDB.Exec("DELETE FROM metric_sample WHERE sampled_at < ?", olderThanMs); tx.Error != nil {
		logrus.Errorf("%v", tx.Error)
		return errors.New(constant.SysError)
	}
	return nil
}

func DashboardPeerSummary(nowMs int64) (vo.DashboardPeerVo, error) {
	result := vo.DashboardPeerVo{}
	type row struct {
		Total    int64
		Enabled  int64
		Disabled int64
		Expired  int64
	}
	var r row
	if tx := sqliteDB.Raw(`SELECT
		COUNT(1) AS total,
		COALESCE(SUM(CASE WHEN disabled = 0 THEN 1 ELSE 0 END),0) AS enabled,
		COALESCE(SUM(CASE WHEN disabled = 1 THEN 1 ELSE 0 END),0) AS disabled,
		COALESCE(SUM(CASE WHEN expires_at > 0 AND expires_at < ? THEN 1 ELSE 0 END),0) AS expired
		FROM peer`, nowMs).Scan(&r); tx.Error != nil {
		logrus.Errorf("%v", tx.Error)
		return result, errors.New(constant.SysError)
	}
	result.Total = r.Total
	result.Enabled = r.Enabled
	result.Disabled = r.Disabled
	result.Expired = r.Expired
	return result, nil
}

func DashboardTrafficSummary() (vo.DashboardTrafficVo, error) {
	result := vo.DashboardTrafficVo{}
	type row struct {
		Download int64
		Upload   int64
	}
	var r row
	if tx := sqliteDB.Raw(`SELECT
		COALESCE(SUM(download_bytes),0) AS download,
		COALESCE(SUM(upload_bytes),0) AS upload
		FROM peer`).Scan(&r); tx.Error != nil {
		logrus.Errorf("%v", tx.Error)
		return result, errors.New(constant.SysError)
	}
	result.DownloadBytes = r.Download
	result.UploadBytes = r.Upload
	result.TotalBytes = r.Download + r.Upload
	result.SinceResetDownloadBytes = r.Download
	result.SinceResetUploadBytes = r.Upload

	nowTime := time.Now()
	dayStart := time.Date(nowTime.Year(), nowTime.Month(), nowTime.Day(), 0, 0, 0, 0, nowTime.Location()).UnixMilli()
	var today row
	if tx := sqliteDB.Raw(`SELECT
		COALESCE(SUM(rx_bytes),0) AS download,
		COALESCE(SUM(tx_bytes),0) AS upload
		FROM traffic_sample WHERE sampled_at >= ?`, dayStart).Scan(&today); tx.Error == nil {
		result.TodayDownloadBytes = today.Download
		result.TodayUploadBytes = today.Upload
	}
	return result, nil
}

func DashboardTopPeers(fromMs int64, toMs int64, limit int) ([]vo.DashboardTopPeerVo, error) {
	if limit <= 0 {
		limit = 10
	}
	rows := make([]vo.DashboardTopPeerVo, 0)
	if tx := sqliteDB.Raw(`SELECT
		p.id AS peer_id,
		p.name AS name,
		p.remark AS remark,
		COALESCE(SUM(ts.rx_bytes),0) AS download,
		COALESCE(SUM(ts.tx_bytes),0) AS upload,
		COALESCE(SUM(ts.rx_bytes + ts.tx_bytes),0) AS total
		FROM traffic_sample ts
		JOIN peer p ON p.id = ts.peer_id
		WHERE ts.sampled_at BETWEEN ? AND ?
		GROUP BY p.id, p.name, p.remark
		ORDER BY total DESC
		LIMIT ?`, fromMs, toMs, limit).Scan(&rows); tx.Error != nil {
		logrus.Errorf("%v", tx.Error)
		return rows, errors.New(constant.SysError)
	}
	return rows, nil
}

func DashboardTrafficTimeseries(fromMs int64, toMs int64, bucketMs int64, source string) ([]vo.DashboardSeriesPointVo, error) {
	rows := make([]vo.DashboardSeriesPointVo, 0)
	if toMs < fromMs {
		return rows, nil
	}
	if bucketMs <= 0 {
		bucketMs = int64(time.Minute / time.Millisecond)
	}

	sourceTable := "traffic_sample"
	fromColumn := "sampled_at"
	switch strings.TrimSpace(strings.ToLower(source)) {
	case "hourly":
		sourceTable = "traffic_aggregate_hourly"
		fromColumn = "hour_start"
	default:
		sourceTable = "traffic_sample"
		fromColumn = "sampled_at"
	}

	alignedFrom := fromMs - (fromMs % bucketMs)
	if tx := sqliteDB.Raw(`SELECT
		(? + CAST((`+fromColumn+` - ?) / ? AS INTEGER) * ?) AS ts,
		COALESCE(SUM(rx_bytes),0) AS download,
		COALESCE(SUM(tx_bytes),0) AS upload
		FROM `+sourceTable+`
		WHERE `+fromColumn+` BETWEEN ? AND ?
		GROUP BY ts
		ORDER BY ts ASC`, alignedFrom, alignedFrom, bucketMs, bucketMs, fromMs, toMs).Scan(&rows); tx.Error != nil {
		logrus.Errorf("%v", tx.Error)
		return rows, errors.New(constant.SysError)
	}

	rowMap := make(map[int64]vo.DashboardSeriesPointVo, len(rows))
	for _, item := range rows {
		rowMap[item.Ts] = item
	}
	filled := make([]vo.DashboardSeriesPointVo, 0, int((toMs-alignedFrom)/bucketMs)+1)
	for ts := alignedFrom; ts <= toMs; ts += bucketMs {
		if item, ok := rowMap[ts]; ok {
			filled = append(filled, item)
			continue
		}
		filled = append(filled, vo.DashboardSeriesPointVo{Ts: ts, Download: 0, Upload: 0})
	}
	return filled, nil
}

func DashboardSystemTimeseries(fromMs int64, toMs int64, bucketMs int64) ([]vo.DashboardSeriesPointVo, error) {
	rows := make([]vo.DashboardSeriesPointVo, 0)
	if !tableExists("metric_sample") {
		return rows, nil
	}
	if toMs < fromMs {
		return rows, nil
	}
	if bucketMs <= 0 {
		bucketMs = int64(time.Minute / time.Millisecond)
	}
	alignedFrom := fromMs - (fromMs % bucketMs)
	if tx := sqliteDB.Raw(`SELECT
		(? + CAST((sampled_at - ?) / ? AS INTEGER) * ?) AS ts,
		AVG(cpu_percent) AS cpu,
		AVG(mem_percent) AS mem
		FROM metric_sample
		WHERE sampled_at BETWEEN ? AND ?
		GROUP BY ts
		ORDER BY ts ASC`, alignedFrom, alignedFrom, bucketMs, bucketMs, fromMs, toMs).Scan(&rows); tx.Error != nil {
		logrus.Errorf("%v", tx.Error)
		return rows, errors.New(constant.SysError)
	}
	return rows, nil
}