package operator_test

import (
	"context"
	"flux/core"
	"flux/core/operator"
	"testing"
)

type timestampEvent struct {
	EventTime int64
}

func TestWatermarkGeneratorPreservesAssignedEventTimestamp(t *testing.T) {
	ctx := context.Background()

	op := operator.NewWatermarkGeneratorOperator[timestampEvent](0)

	if err := op.Open(ctx); err != nil {
		t.Fatalf("open operator failed: %v", err)
	}

	t.Cleanup(func() {
		if err := op.Close(ctx); err != nil {
			t.Errorf("close operator: %v", err)
		}
	})

	out := &captureEmitter[timestampEvent]{}
	input := core.NewDataRecord("user-1", timestampEvent{EventTime: 999}, 40)

	if err := op.Process(ctx, input, out); err != nil {
		t.Fatalf("process operator failed: %v", err)
	}

	records := out.Records()
	if len(records) == 0 {
		t.Fatal("operator emitted no records")
	}

	got := records[0]

	if got.Type != core.RecordData {
		t.Fatalf("first output type: got %v, want RecordData", got.Type)
	}

	if got.Key != "user-1" {
		t.Errorf("data key: got %q, want %q", got.Key, "user-1")
	}

	if got.Value != input.Value {
		t.Errorf("data value: got %+v, want %+v", got.Value, input.Value)
	}

	if got.Timestamp != 40 {
		t.Errorf("data timestamp: got %d, want %d", got.Timestamp, 40)
	}
}