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/window.go | 110 ++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 110 insertions(+)
 create mode 100644 core/operator/window.go

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

diff --git a/core/operator/window.go b/core/operator/window.go
new file mode 100644
index 0000000..610f6e3
--- /dev/null
+++ b/core/operator/window.go
@@ -0,0 +1,110 @@
+package operator
+
+import (
+	"context"
+	"fmt"
+
+	"flux/core"
+)
+
+// WindowFunction is for user interface implements the behaviour
+type WindowFunction[IN, OUT any] interface {
+	Apply(ctx context.Context, key string, window core.Window, in []IN, out Emitter[OUT]) error
+}
+
+type WindowOperator[IN, OUT any] struct {
+	assigner    core.WindowAssigner[IN]
+	userfunc    WindowFunction[IN, OUT]
+	backend     core.StateBackend
+	timeSerivce core.InternalTimerService
+
+	windowState core.ValueState[[]IN]
+}
+
+func NewWindowOperator[IN, OUT any](assigner core.WindowAssigner[IN], userfunc WindowFunction[IN, OUT]) *WindowOperator[IN, OUT] {
+	return &WindowOperator[IN, OUT]{
+		assigner:    assigner,
+		userfunc:    userfunc,
+		timeSerivce: core.NewTimerService(),
+	}
+}
+
+func (o *WindowOperator[IN, OUT]) Open(ctx context.Context) error {
+	be, ok := core.ExtractStateBackend(ctx)
+	if !ok {
+		return fmt.Errorf("cannot extract state backend, got: %v", be)
+	}
+	o.backend = be
+	windowState, err := be.GetValueState(ctx, "window-elements")
+	if err != nil {
+		return fmt.Errorf("cannot get value state window-elements error: %w", err)
+	}
+
+	o.windowState = core.CastValueType[[]IN](windowState)
+
+	return nil
+}
+
+func (o *WindowOperator[IN, OUT]) Process(ctx context.Context, record core.StreamRecord[IN], out Emitter[OUT]) error {
+	switch record.Type {
+	case core.RecordData:
+		// current windows state will hold all record in window
+		// assign and get list window
+		windows := o.assigner.AssignWindows(record.Value, record.Timestamp)
+		for _, w := range windows {
+			o.backend.SetCurrentNamespace(w)
+			o.backend.SetCurrentKey(record.Key)
+			o.timeSerivce.Register(w.MaxTimestamp(), record.Key, w)
+
+			currentWindowState, err := o.windowState.Value(ctx)
+			if err != nil {
+				return err
+			}
+
+			currentWindowState = append(currentWindowState, record.Value)
+
+			o.windowState.Update(ctx, currentWindowState)
+		}
+
+		return nil
+	case core.RecordWatermark:
+		deadlines := o.timeSerivce.AdvanceWatermark(record.Timestamp)
+
+		for _, timer := range deadlines {
+			o.backend.SetCurrentKey(timer.Key)
+			o.backend.SetCurrentNamespace(timer.Namespace)
+
+			currentWindowState, err := o.windowState.Value(ctx)
+			if err != nil {
+				return err
+			}
+
+			// NOTE: this is workaround create inline window, refactor later
+
+			if err := o.userfunc.Apply(ctx, timer.Key, timer.Namespace.(core.Window), currentWindowState, out); err != nil {
+				return err
+			}
+
+			if err := o.windowState.Clear(ctx); err != nil {
+				return err
+			}
+
+		}
+
+		return out.Emit(ctx, core.StreamRecord[OUT]{
+			Timestamp: record.Timestamp,
+			Type:      core.RecordWatermark,
+		})
+	default:
+		return out.Emit(ctx, core.StreamRecord[OUT]{
+			BarrierID: record.BarrierID,
+			Timestamp: record.Timestamp,
+			Key:       record.Key,
+			Type:      record.Type,
+		})
+	}
+}
+
+func (o *WindowOperator[IN, OUT]) Close(ctx context.Context) error {
+	return nil
+}
-- 
cgit v1.3.1