From 96c07bab1086510a55a004e9a1b90c30bc723a5c Mon Sep 17 00:00:00 2001
From: sangtran <tranquangsang12.7@gmail.com>
Date: Sat, 29 Aug 2026 01:08:28 +0700
Subject: feat: init base draft support window

---
 core/operator/watermark.go | 61 ++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 61 insertions(+)
 create mode 100644 core/operator/watermark.go

(limited to 'core/operator/watermark.go')

diff --git a/core/operator/watermark.go b/core/operator/watermark.go
new file mode 100644
index 0000000..3d591e5
--- /dev/null
+++ b/core/operator/watermark.go
@@ -0,0 +1,61 @@
+package operator
+
+import (
+	"context"
+
+	"flux/core"
+)
+
+// TimestampAssigner is responsible for event time defined by user
+type TimestampAssigner[IN any] func(in IN) int64
+
+type WaterGeneratorOperator[IN any] struct {
+	assigner             TimestampAssigner[IN]
+	maxOutOfOrder        int64
+	currentMaxTimestamp  int64
+	lastEmittedWatermark int64
+}
+
+func NewWatermarkGeneratorOperator[IN any](maxOutOfOrder int64, assigner TimestampAssigner[IN]) *WaterGeneratorOperator[IN] {
+	return &WaterGeneratorOperator[IN]{
+		maxOutOfOrder:        maxOutOfOrder,
+		assigner:             assigner,
+		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 := w.assigner(record.Value)
+		record.Timestamp = eventTime
+
+		// 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
+}
-- 
cgit v1.3.1