package main

import (
	"sync"
	"time"

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

type LakeIngesterStatsRecorder struct {
	ingestedBytes *timeseries.TimeSeries
	ingestedRows  *timeseries.TimeSeries
	mu            sync.Mutex
}

type LakeIngesterStatsData struct {
	BytesPerSecond float64
	RowsPerSecond  float64
}

func NewLakeIntesterStatsRecorder() *LakeIngesterStatsRecorder {
	ingestedBytes, err := timeseries.NewTimeSeries()
	if err != nil {
		panic(err)
	}
	ingestedRows, err := timeseries.NewTimeSeries()
	if err != nil {
		panic(err)
	}
	return &LakeIngesterStatsRecorder{
		ingestedBytes: ingestedBytes,
		ingestedRows:  ingestedRows,
	}
}

func (stats *LakeIngesterStatsRecorder) RecordMetric(bytes int, rows int) {
	stats.mu.Lock()
	defer stats.mu.Unlock()
	stats.ingestedBytes.Increase(bytes)
	stats.ingestedRows.Increase(rows)
}

func (stats *LakeIngesterStatsRecorder) Stats(statsWindow time.Duration) LakeIngesterStatsData {
	stats.mu.Lock()
	defer stats.mu.Unlock()

	bytesPerSecond := stats.calcPerSecond(stats.ingestedBytes, statsWindow)
	rowsPerSecond := stats.calcPerSecond(stats.ingestedRows, statsWindow)
	return LakeIngesterStatsData{
		BytesPerSecond: bytesPerSecond,
		RowsPerSecond:  rowsPerSecond,
	}
}

func (stats *LakeIngesterStatsRecorder) 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()
}