package main

import (
	"sync"
	"time"

	timeseries "github.com/codesuki/go-time-series"
)

type LakeConsumeStatsRecorder struct {
	consumedBytes *timeseries.TimeSeries
	consumedRows  *timeseries.TimeSeries
	mu            sync.Mutex
}

type LakeConsumeStatsData struct {
	BytesPerSecond float64
	RowsPerSecond  float64
}

func NewLakeConsumeStatsRecorder() *LakeConsumeStatsRecorder {
	consumedBytes, err := timeseries.NewTimeSeries()
	if err != nil {
		panic(err)
	}
	consumedRows, err := timeseries.NewTimeSeries()
	if err != nil {
		panic(err)
	}
	return &LakeConsumeStatsRecorder{
		consumedBytes: consumedBytes,
		consumedRows:  consumedRows,
	}
}

func (stats *LakeConsumeStatsRecorder) RecordMetric(bytes int, rows int) {
	stats.mu.Lock()
	defer stats.mu.Unlock()
	stats.consumedBytes.Increase(bytes)
	stats.consumedRows.Increase(rows)
}

func (stats *LakeConsumeStatsRecorder) Stats(statsWindow time.Duration) LakeConsumeStatsData {
	stats.mu.Lock()
	defer stats.mu.Unlock()

	bytesPerSecond := stats.calcPerSecond(stats.consumedBytes, statsWindow)
	rowsPerSecond := stats.calcPerSecond(stats.consumedRows, statsWindow)
	return LakeConsumeStatsData{
		BytesPerSecond: bytesPerSecond,
		RowsPerSecond:  rowsPerSecond,
	}
}

func (stats *LakeConsumeStatsRecorder) calcPerSecond(ts *timeseries.TimeSeries, duration time.Duration) float64 {
	amount, err := ts.Range(time.Now().Add(-duration), time.Now())
	if err != nil {
		return -1
	}

	return float64(amount) / duration.Seconds()
}