package operator

import (
	"context"

	"flux/core"
)

type WaterGeneratorOperator[IN any] struct {
	maxOutOfOrder        int64
	currentMaxTimestamp  int64
	lastEmittedWatermark int64
}

func NewWatermarkGeneratorOperator[IN any](maxOutOfOrder int64) *WaterGeneratorOperator[IN] {
	return &WaterGeneratorOperator[IN]{
		maxOutOfOrder:        maxOutOfOrder,
		lastEmittedWatermark: -1,
	}
}

func (w *WaterGeneratorOperator[IN]) Open(ctx context.Context) error {
	return nil
}

func (w *WaterGeneratorOperator[IN]) Process(ctx context.Context, record core.StreamRecord[IN], out Emitter[IN]) error {
	if record.Type == core.RecordData {
		if err := out.Emit(ctx, record); err != nil {
			return err
		}
		// query the time
		eventTime := record.Timestamp

		// update the max timestamp
		if w.currentMaxTimestamp < eventTime {
			w.currentMaxTimestamp = eventTime
		}

		// calc the watermark

		watermark := w.currentMaxTimestamp - w.maxOutOfOrder

		// check if current watermark later than previous watermark
		if watermark > w.lastEmittedWatermark {
			w.lastEmittedWatermark = watermark
			return out.Emit(ctx, core.NewWatermarkRecord[IN](watermark))
		}
		return nil
	}
	return out.Emit(ctx, record)
}

func (w *WaterGeneratorOperator[IN]) Close(ctx context.Context) error {
	return nil
}