package prompush

import (
	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/common"
	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/relabel"
	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prommetadata"
	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
	"github.com/VictoriaMetrics/metrics"
)

var (
	rowsInserted         = metrics.NewCounter(`vm_rows_inserted_total{type="promscrape"}`)
	rowsPerInsert        = metrics.NewHistogram(`vm_rows_per_insert{type="promscrape"}`)
	metadataRowsInserted = metrics.NewCounter(`vm_metadata_rows_inserted_total{type="promscrape"}`)
)

const maxRowsPerBlock = 10000

// Push pushes wr for the given at to storage.
func Push(wr *prompb.WriteRequest) {
	ctx := common.GetInsertCtx()
	defer common.PutInsertCtx(ctx)

	tss := wr.Timeseries
	for len(tss) > 0 {
		// Process big tss in smaller blocks in order to reduce maximum memory usage
		samplesCount := 0
		i := 0
		for i < len(tss) {
			samplesCount += len(tss[i].Samples)
			i++
			if samplesCount > maxRowsPerBlock {
				break
			}
		}
		tssBlock := tss
		if i < len(tss) {
			tssBlock = tss[:i]
			tss = tss[i:]
		} else {
			tss = nil
		}
		push(ctx, tssBlock)
	}
	if prommetadata.IsEnabled() {
		if err := ctx.WriteMetadata(wr.Metadata); err != nil {
			logger.Errorf("cannot write promscrape metrics metadata to storage: %s", err)
		} else {
			metadataRowsInserted.Add(len(wr.Metadata))
		}
	}
}

func push(ctx *common.InsertCtx, tss []prompb.TimeSeries) {
	rowsLen := 0
	for i := range tss {
		rowsLen += len(tss[i].Samples)
	}
	ctx.Reset(rowsLen)
	rowsTotal := 0
	hasRelabeling := relabel.HasRelabeling()
	for i := range tss {
		ts := &tss[i]
		rowsTotal += len(ts.Samples)
		ctx.Labels = ctx.Labels[:0]
		for j := range ts.Labels {
			label := &ts.Labels[j]
			ctx.AddLabel(label.Name, label.Value)
		}
		if !ctx.TryPrepareLabels(hasRelabeling) {
			continue
		}
		var metricNameRaw []byte
		var err error
		for i := range ts.Samples {
			r := &ts.Samples[i]
			metricNameRaw, err = ctx.WriteDataPointExt(metricNameRaw, ctx.Labels, r.Timestamp, r.Value)
			if err != nil {
				logger.Errorf("cannot write promscrape data to storage: %s", err)
				return
			}
		}
	}
	rowsInserted.Add(rowsTotal)
	rowsPerInsert.Update(float64(rowsTotal))
	if err := ctx.FlushBufs(); err != nil {
		logger.Errorf("cannot flush promscrape data to storage: %s", err)
	}
}
