Verified feat: init base draft support window

23 files changed, 1123 insertions, 0 deletions

diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..68729d6
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,2 @@
+.idea/
+.DS_Store
\ No newline at end of file
diff --git a/README.md b/README.md
new file mode 100644
index 0000000..d947411
--- /dev/null
+++ b/README.md
@@ -0,0 +1,14 @@
+# flux
+
+
+
+
+> [!NOTE]
+> Namespace should be interface instead of any/interface{} because it cost boxing & GC heavy things and also losing the type safety
+> 
+
+
+
+### Namespace improvement
+Should serialize to raw []byte to archive serialize speed, flexible, zero GC(with buffer bool) also compatible with PebbleDB
+
diff --git a/core/hash.go b/core/hash.go
new file mode 100644
index 0000000..af15546
--- /dev/null
+++ b/core/hash.go
@@ -0,0 +1,19 @@
+package core
+
+import "hash/fnv"
+
+const DefaultMaxParallelism = 128
+
+func AssignToKeyGroup(key string, maxParallelism int) int {
+
+        if maxParallelism <= 0 {
+                maxParallelism = DefaultMaxParallelism
+        }
+
+        // hash
+        h := fnv.New32a()
+        h.Write([]byte(key))
+        hash := h.Sum32()
+
+        return int(hash % uint32(maxParallelism))
+}
diff --git a/core/operator/filter.go b/core/operator/filter.go
new file mode 100644
index 0000000..80bbb77
--- /dev/null
+++ b/core/operator/filter.go
@@ -0,0 +1,36 @@
+package operator
+
+import (
+        "context"
+        "flux/core"
+)
+
+type FilterFunc[IN any] func(in IN) bool
+
+type FilterOperator[IN any] struct {
+        userFunc FilterFunc[IN]
+}
+
+func NewFilterOperator[IN any](userFunc FilterFunc[IN]) *FilterOperator[IN] {
+        return &FilterOperator[IN]{userFunc: userFunc}
+}
+
+func (f *FilterOperator[IN]) Open(ctx context.Context) error {
+        return nil
+}
+
+func (f *FilterOperator[IN]) Process(ctx context.Context, record core.StreamRecord[IN], out Emitter[IN]) error {
+        if record.Type != core.RecordData {
+                return out.Emit(ctx, record)
+        }
+
+        if f.userFunc(record.Value) {
+                return out.Emit(ctx, record)
+        }
+
+        return nil
+}
+
+func (f *FilterOperator[IN]) Close(ctx context.Context) error {
+        return nil
+}
diff --git a/core/operator/keyby.go b/core/operator/keyby.go
new file mode 100644
index 0000000..56ef34e
--- /dev/null
+++ b/core/operator/keyby.go
@@ -0,0 +1,32 @@
+package operator
+
+import (
+        "context"
+        "flux/core"
+)
+
+type KeySelector[IN any] func(in IN) string
+type KeyByOperator[IN any] struct {
+        selector KeySelector[IN]
+}
+
+func NewKeyByOperator[IN any](selector KeySelector[IN]) *KeyByOperator[IN] {
+        return &KeyByOperator[IN]{selector: selector}
+}
+
+func (k *KeyByOperator[IN]) Open(ctx context.Context) error {
+        return nil
+}
+
+func (k *KeyByOperator[IN]) Process(ctx context.Context, record core.StreamRecord[IN], out Emitter[IN]) error {
+        if record.Type != core.RecordData {
+                return out.Emit(ctx, record)
+        }
+
+        record.Key = k.selector(record.Value)
+        return out.Emit(ctx, record)
+}
+
+func (k *KeyByOperator[IN]) Close(ctx context.Context) error {
+        return nil
+}
diff --git a/core/operator/keyed_process.go b/core/operator/keyed_process.go
new file mode 100644
index 0000000..1d3c86c
--- /dev/null
+++ b/core/operator/keyed_process.go
@@ -0,0 +1,81 @@
+package operator
+
+import (
+        "context"
+        "flux/core"
+        "fmt"
+)
+
+type KeyedProcesser[IN, OUT any] interface {
+        Open(ctx context.Context) error
+        ProcessElement(ctx context.Context, record core.StreamRecord[IN], out Emitter[OUT]) error
+        Close(ctx context.Context) error
+        OnTimer(ctx context.Context, timestamp int64, out Emitter[OUT]) error
+}
+
+type KeyProcessOperator[IN, OUT any] struct {
+        userFunc    KeyedProcesser[IN, OUT]
+        timeService core.InternalTimerService
+        backend     core.StateBackend
+}
+
+func NewKeyProcessOperator[IN, OUT any](userFunc KeyedProcesser[IN, OUT]) *KeyProcessOperator[IN, OUT] {
+        return &KeyProcessOperator[IN, OUT]{
+                userFunc:    userFunc,
+                timeService: core.NewTimerService(),
+        }
+}
+
+func (k *KeyProcessOperator[IN, OUT]) Open(ctx context.Context) error {
+        // init backend
+        be, ok := core.ExtractStateBackend(ctx)
+        if !ok {
+                return fmt.Errorf("state backend not found")
+        }
+
+        k.backend = be
+        // inject the timer to context for user use
+        ctx = core.InjectTimerService(ctx, k.timeService)
+        return k.userFunc.Open(ctx)
+
+}
+
+func (k *KeyProcessOperator[IN, OUT]) Process(ctx context.Context, record core.StreamRecord[IN], out Emitter[OUT]) error {
+
+        switch record.Type {
+        case core.RecordData:
+                if err := k.backend.SetCurrentKey(record.Key); err != nil {
+                        return err
+                }
+
+                return k.userFunc.ProcessElement(ctx, record, out)
+        case core.RecordWatermark:
+                // advance the timer
+                triggerTimerQueue := k.timeService.AdvanceWatermark(record.Timestamp)
+
+                for _, t := range triggerTimerQueue {
+
+                        if err := k.backend.SetCurrentKey(t.Key); err != nil {
+                                return err
+                        }
+                        if err := k.userFunc.OnTimer(ctx, t.Timestamp, out); err != nil {
+                                return err
+                        }
+                }
+                return out.Emit(ctx, core.StreamRecord[OUT]{
+                        Type:      core.RecordWatermark,
+                        Timestamp: record.Timestamp,
+                })
+        default:
+                return out.Emit(ctx, core.StreamRecord[OUT]{
+                        Timestamp: record.Timestamp,
+                        BarrierID: record.BarrierID,
+                        Key:       record.Key,
+                })
+        }
+
+}
+
+func (k *KeyProcessOperator[IN, OUT]) Close(ctx context.Context) error {
+        return k.userFunc.Close(ctx)
+}
diff --git a/core/operator/map.go b/core/operator/map.go
new file mode 100644
index 0000000..2165c25
--- /dev/null
+++ b/core/operator/map.go
@@ -0,0 +1,53 @@
+package operator
+
+import (
+        "context"
+        "flux/core"
+        "fmt"
+)
+
+// MapFunc is business logic for user implement
+type MapFunc[IN, OUT any] func(in IN) (OUT, error)
+
+type MapOperator[IN, OUT any] struct {
+        userFunc MapFunc[IN, OUT]
+}
+
+func NewMapOperator[IN, OUT any](fn MapFunc[IN, OUT]) *MapOperator[IN, OUT] {
+        return &MapOperator[IN, OUT]{
+                userFunc: fn,
+        }
+}
+
+func (fo *MapOperator[IN, OUT]) Open(ctx context.Context) error {
+        return nil
+}
+
+func (fo *MapOperator[IN, OUT]) Process(ctx context.Context, record core.StreamRecord[IN], out Emitter[OUT]) error {
+        if record.Type != core.RecordData {
+                //       watermark and barrier
+                return out.Emit(ctx, core.StreamRecord[OUT]{
+                        Type:      record.Type,
+                        Key:       record.Key,
+                        Timestamp: record.Timestamp,
+                        BarrierID: record.BarrierID,
+                })
+        }
+
+        outValue, err := fo.userFunc(record.Value)
+
+        if err != nil {
+                return fmt.Errorf("cannot process value %v: %w", record.Value, err)
+        }
+
+        return out.Emit(ctx, core.StreamRecord[OUT]{
+                Type:      core.RecordData,
+                Key:       record.Key,
+                Timestamp: record.Timestamp,
+                Value:     outValue,
+        })
+}
+
+func (fo *MapOperator[IN, OUT]) Close(ctx context.Context) error {
+        return nil
+}
diff --git a/core/operator/operator.go b/core/operator/operator.go
new file mode 100644
index 0000000..0c588e5
--- /dev/null
+++ b/core/operator/operator.go
@@ -0,0 +1,16 @@
+package operator
+
+import (
+        "context"
+        "flux/core"
+)
+
+type Emitter[OUT any] interface {
+        Emit(ctx context.Context, record core.StreamRecord[OUT]) error
+}
+
+type Operator[IN, OUT any] interface {
+        Open(ctx context.Context) error
+        Process(ctx context.Context, record core.StreamRecord[IN], out Emitter[OUT]) error
+        Close(ctx context.Context) error
+}
diff --git a/core/operator/stateful_map.go b/core/operator/stateful_map.go
new file mode 100644
index 0000000..295c656
--- /dev/null
+++ b/core/operator/stateful_map.go
@@ -0,0 +1,75 @@
+package operator
+
+import (
+        "context"
+        "fmt"
+
+        "flux/core"
+)
+
+// RichMapFunc is a stateful map on the user side: Open grabs the state handles
+// off ctx, Map runs per record with the backend already pointed at that
+// record's key.
+type RichMapFunc[IN, OUT any] interface {
+        Open(ctx context.Context) error
+        Map(ctx context.Context, in IN) (OUT, error)
+        Close(ctx context.Context) error
+}
+
+type StatefulMapOperator[IN, OUT any] struct {
+        userFunc RichMapFunc[IN, OUT]
+        backend  core.StateBackend
+}
+
+func NewStatefulMapOperator[IN, OUT any](fn RichMapFunc[IN, OUT]) *StatefulMapOperator[IN, OUT] {
+        return &StatefulMapOperator[IN, OUT]{userFunc: fn}
+}
+
+func (s *StatefulMapOperator[IN, OUT]) Open(ctx context.Context) error {
+        // Get backend state from context via using context KV
+        be, ok := core.ExtractStateBackend(ctx)
+        if !ok || be == nil {
+                return fmt.Errorf("stateful map: no state backend in context")
+        }
+
+        s.backend = be
+        // The user func opens its ValueState off the same ctx.
+        return s.userFunc.Open(ctx)
+}
+
+func (s *StatefulMapOperator[IN, OUT]) Process(ctx context.Context, record core.StreamRecord[IN], out Emitter[OUT]) error {
+        if record.Type != core.RecordData {
+                return out.Emit(ctx, core.StreamRecord[OUT]{
+                        Type:      record.Type,
+                        Timestamp: record.Timestamp,
+                        Key:       record.Key,
+                        BarrierID: record.BarrierID,
+                })
+        }
+
+        // Without a key every record would silently share the "" state slot.
+        if record.Key == "" {
+                return fmt.Errorf("stateful map: record %v has no key, keyBy it first", record.Value)
+        }
+
+        if err := s.backend.SetCurrentKey(record.Key); err != nil {
+                return fmt.Errorf("set current key %q: %w", record.Key, err)
+        }
+
+        // run the user transform
+        res, err := s.userFunc.Map(ctx, record.Value)
+        if err != nil {
+                return fmt.Errorf("cannot process value %v: %w", record.Value, err)
+        }
+
+        return out.Emit(ctx, core.StreamRecord[OUT]{
+                Type:      core.RecordData,
+                Timestamp: record.Timestamp,
+                Key:       record.Key,
+                Value:     res,
+        })
+}
+
+func (s *StatefulMapOperator[IN, OUT]) Close(ctx context.Context) error {
+        return s.userFunc.Close(ctx)
+}
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
+}
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
+}
diff --git a/core/record.go b/core/record.go
new file mode 100644
index 0000000..e18e58d
--- /dev/null
+++ b/core/record.go
@@ -0,0 +1,40 @@
+package core
+
+type RecordType byte
+
+const (
+        RecordData RecordType = iota
+        RecordWatermark
+        RecordBarrier
+)
+
+type StreamRecord[T any] struct {
+        Value     T
+        Key       string // for keyBy stateful processing
+        Timestamp int64  // epoch ms time
+        BarrierID uint64 // ID checkpoint barrier
+        Type      RecordType
+}
+
+func NewDataRecord[T any](key string, val T, timestamp int64) StreamRecord[T] {
+        return StreamRecord[T]{
+                Value:     val,
+                Key:       key,
+                Timestamp: timestamp,
+                Type:      RecordData,
+        }
+}
+
+func NewWatermarkRecord[T any](timestamp int64) StreamRecord[T] {
+        return StreamRecord[T]{
+                Type:      RecordWatermark,
+                Timestamp: timestamp,
+        }
+}
+
+func NewBarrierRecord[T any](barrierID uint64) StreamRecord[T] {
+        return StreamRecord[T]{
+                Type:      RecordBarrier,
+                BarrierID: barrierID,
+        }
+}
diff --git a/core/state.go b/core/state.go
new file mode 100644
index 0000000..d6d7847
--- /dev/null
+++ b/core/state.go
@@ -0,0 +1,10 @@
+package core
+
+import "context"
+
+// ValueState is for client side using to update state
+type ValueState[T any] interface {
+        Value(ctx context.Context) (T, error)
+        Update(ctx context.Context, newVal T) (T, error)
+        Clear(ctx context.Context) error
+}
diff --git a/core/state_backend.go b/core/state_backend.go
new file mode 100644
index 0000000..9f9f397
--- /dev/null
+++ b/core/state_backend.go
@@ -0,0 +1,62 @@
+package core
+
+import (
+        "context"
+)
+
+type StateBackend interface {
+        // For Framework ===
+        // SetCurrentKey will perform in Memory
+        SetCurrentKey(key string) error
+        // GetCurrentKey will perform in Memory
+        GetCurrentKey() string
+        SetCurrentNamespace(namespace any) error
+        GetCurrentNamespace() any
+
+        // For User ===
+        // GetValueState will perform in disk
+        GetValueState(ctx context.Context, key string) (ValueState[any], error)
+}
+
+type stateBackendKeyType struct{}
+
+var stateBackendKey stateBackendKeyType
+
+type typedValueState[T any] struct {
+        inner ValueState[any]
+}
+
+func CastValueType[T any](inner ValueState[any]) ValueState[T] {
+        return &typedValueState[T]{inner: inner}
+}
+
+func (s *typedValueState[T]) Value(ctx context.Context) (T, error) {
+        val, err := s.inner.Value(ctx)
+        if err != nil {
+                var zero T
+                return zero, err
+        }
+        if val == nil {
+                var zero T
+                return zero, nil
+        }
+        return val.(T), nil
+}
+
+func (s *typedValueState[T]) Update(ctx context.Context, newVal T) (T, error) {
+        _, err := s.inner.Update(ctx, newVal)
+        return newVal, err
+}
+
+func (s *typedValueState[T]) Clear(ctx context.Context) error {
+        return s.inner.Clear(ctx)
+}
+
+func InjectStateBackend(ctx context.Context, state StateBackend) context.Context {
+        return context.WithValue(ctx, stateBackendKey, state)
+}
+
+func ExtractStateBackend(ctx context.Context) (StateBackend, bool) {
+        backend, ok := ctx.Value(stateBackendKey).(StateBackend)
+        return backend, ok
+}
diff --git a/core/timer.go b/core/timer.go
new file mode 100644
index 0000000..3d17299
--- /dev/null
+++ b/core/timer.go
@@ -0,0 +1,113 @@
+package core
+
+import (
+        "container/heap"
+        "context"
+)
+
+// If using rockDB no need priority queue anymore
+// implement priority queue from golang
+// ref: https://pkg.go.dev/container/heap#example-package-PriorityQueue
+
+type timerServiceKey struct{}
+
+// InjectTimerService inject the timer into Context
+func InjectTimerService(ctx context.Context, ts TimerService) context.Context {
+        return context.WithValue(ctx, timerServiceKey{}, ts)
+}
+
+// ExtractTimerService get the timer for user from context
+func ExtractTimerService(ctx context.Context) (TimerService, bool) {
+        ts, ok := ctx.Value(timerServiceKey{}).(TimerService)
+        return ts, ok
+}
+
+type Timer struct {
+        // Timestamp is the time when Timer is triggered
+        Timestamp int64
+        // Key is for who User involve
+        Key string
+        // Window is for time range start-end of
+        // String actually cost more bytes than Window(16bytes) 
+        // and we dont want to decode from namespace to Window  
+        Namespace any 
+}
+
+// Priority Queue (Min Heap). Smallest timestamp will be on top
+
+type timerHeap []Timer
+
+func (h *timerHeap) Len() int           { return len(*h) }
+func (h *timerHeap) Less(i, j int) bool { return (*h)[i].Timestamp < (*h)[j].Timestamp }
+func (h *timerHeap) Swap(i, j int) {
+        (*h)[i], (*h)[j] = (*h)[j], (*h)[i]
+}
+
+func (h *timerHeap) Push(x any) {
+        *h = append(*h, x.(Timer))
+}
+
+func (h *timerHeap) Pop() any {
+        old := *h
+        n := len(old)
+        last := old[n-1]
+        *h = old[:n-1]
+        return last
+}
+
+type TimerService interface {
+        Register(time int64, key string, keyspace any)
+        CurrentWatermark() int64
+}
+
+type InternalTimerService interface {
+        TimerService
+        AdvanceWatermark(watermark int64) []Timer
+}
+
+type timerService struct {
+        queue            timerHeap
+        currentWatermark int64
+}
+
+func NewTimerService() InternalTimerService {
+        return &timerService{
+                queue:            make(timerHeap, 0),
+                currentWatermark: -1,
+        }
+}
+
+func (t *timerService) Register(time int64, key string, namespace any) {
+        heap.Push(&t.queue, Timer{
+                Timestamp: time,
+                Key:       key,
+                Namespace: namespace,
+        })
+}
+
+func (t *timerService) CurrentWatermark() int64 {
+        return t.currentWatermark
+}
+
+func (t *timerService) AdvanceWatermark(watermark int64) []Timer {
+        if watermark < t.currentWatermark {
+                // no need to advance when get smaller
+                return nil
+        }
+
+        t.currentWatermark = watermark
+
+        var timers []Timer
+        for t.queue.Len() > 0 {
+                earliestTimer := t.queue[0]
+
+                if earliestTimer.Timestamp > watermark {
+                        break
+                }
+
+                pop := heap.Pop(&t.queue).(Timer)
+                timers = append(timers, pop)
+        }
+
+        return timers
+}
diff --git a/core/window.go b/core/window.go
new file mode 100644
index 0000000..6c4b98f
--- /dev/null
+++ b/core/window.go
@@ -0,0 +1,54 @@
+package core 
+
+import "fmt"
+
+type Window interface {
+        // MaxTimestamp is the maximum time the Window is Open
+        MaxTimestamp() int64
+        // ToString convert Window to string format
+        ToString() string
+}
+
+type TimeWindow struct {
+        Start int64
+        End   int64
+}
+
+func (t TimeWindow) MaxTimestamp() int64 {
+        // MaxTimestamp is 1ms before the End
+        return t.End - 1
+}
+
+func (t TimeWindow) ToString() string {
+        return fmt.Sprintf("timewindow(%d,%d)", t.Start, t.End)
+}
+
+type WindowAssigner[IN any] interface {
+        AssignWindows(element IN, timestamp int64) []Window
+}
+
+type TumblingWindows[IN any] struct {
+        Size int64
+}
+
+func NewTumblingWindows[IN any](size int64) *TumblingWindows[IN] {
+        return &TumblingWindows[IN]{
+                Size: size,
+        }
+}
+
+func (w *TumblingWindows[IN]) AssignWindows(element IN, timestamp int64) []Window {
+        // module to get start end
+        // if timestamp = 1200, size 1000
+        // so start will 1200 - 200 = 1000, end will 1000 + 1000 = 2000
+        start := timestamp - (timestamp % w.Size)
+
+        end := start + w.Size
+
+        return []Window{
+                TimeWindow{
+                        Start: start,
+                        End: end,
+                },
+        }
+}
diff --git a/go.mod b/go.mod
new file mode 100644
index 0000000..1094a64
--- /dev/null
+++ b/go.mod
@@ -0,0 +1,3 @@
+module flux
+
+go 1.26.3
diff --git a/main.go b/main.go
new file mode 100644
index 0000000..1b4a4be
--- /dev/null
+++ b/main.go
@@ -0,0 +1,96 @@
+package main
+
+import (
+        "context"
+        "fmt"
+        "strconv"
+        "sync"
+
+        "flux/core"
+        "flux/core/operator"
+        "flux/runtime"
+        "flux/state"
+        "flux/transport"
+)
+
+type Transaction struct {
+        User      string
+        Amount    int
+        Timestamp int64
+}
+
+type TransactionSummary[IN Transaction, OUT Transaction] struct {
+        totalTransaction int
+        totalAmount      int
+        average          int
+}
+
+func (t *TransactionSummary[IN, OUT]) Apply(ctx context.Context, key string, window core.Window, in []Transaction, out transport.Emitter[Transaction]) error {
+        for _, v := range in {
+                t.totalAmount += v.Amount
+                out.Emit(ctx, core.StreamRecord[Transaction]{
+                        Value: v,
+                        Type: core.RecordData,
+                        Timestamp: v.Timestamp,
+                })
+        }
+
+        return nil
+}
+
+func main() {
+        // init
+        ctx := context.Background()
+
+        backend := state.NewMemoryStateBackend(1)
+
+        ctx = core.InjectStateBackend(ctx, backend)
+
+        txList := []Transaction{
+                {
+                        User:      "Sang",
+                        Amount:    1,
+                        Timestamp: 1000,
+                },
+                {
+                        User:      "Sang",
+                        Amount:    4,
+                        Timestamp: 1200,
+                },
+                {
+                        User:      "Sang",
+                        Amount:    10,
+                        Timestamp: 1600,
+                },
+        }
+
+        ch1 := make(chan core.StreamRecord[Transaction], len(txList))
+        ch2 := make(chan core.StreamRecord[Transaction], len(txList))
+
+        wo := operator.NewWatermarkGeneratorOperator[Transaction](0, func(in Transaction) int64 {
+                return in.Timestamp
+        })
+
+        var wg sync.WaitGroup
+
+        runtime.RunTask[Transaction, Transaction](ctx, transport.NewLocalReceiveChannel(ch1), transport.NewLocalEmitChannel(ch2), wo)
+
+        wg.Go(func() {
+                for i, tx := range txList {
+                        ch1 <- core.NewDataRecord(strconv.Itoa(i), tx, tx.Timestamp)
+                }
+
+                close(ch1)
+        })
+
+        wg.Go(func() {
+                for v := range ch2 {
+                        if v.Type == core.RecordData {
+                                fmt.Println("data: ", v.Value)
+                        } else {
+                                fmt.Println("watermark: ", v.Timestamp)
+                        }
+                }
+        })
+        wg.Wait()
+}
diff --git a/runtime/task.go b/runtime/task.go
new file mode 100644
index 0000000..734976e
--- /dev/null
+++ b/runtime/task.go
@@ -0,0 +1,66 @@
+package runtime
+
+import (
+        "context"
+        "fmt"
+        "time"
+
+        "flux/core/operator"
+        "flux/transport"
+)
+
+// RunTask runs a stream task in the background. The returned channel yields the
+// first error that stopped it (context.Canceled when ctx ends early) and is
+// closed when the task is done, so callers can join on it.
+func RunTask[IN, OUT any](ctx context.Context, input transport.Receiver[IN], output transport.Emitter[OUT], op operator.Operator[IN, OUT]) <-chan error {
+        done := make(chan error, 1)
+
+        go func() {
+                defer close(done)
+                if err := runTask(ctx, input, output, op); err != nil {
+                        done <- err
+                }
+        }()
+
+        return done
+}
+
+func runTask[IN, OUT any](ctx context.Context, input transport.Receiver[IN], output transport.Emitter[OUT], op operator.Operator[IN, OUT]) (err error) {
+        // ponytail: assumes one writer per output channel. Fan-in needs a refcount
+        // or a dedicated closer, otherwise the second Close panics.
+        defer func() {
+                if cerr := output.Close(); cerr != nil && err == nil {
+                        err = fmt.Errorf("close output: %w", cerr)
+                }
+        }()
+
+        if err := op.Open(ctx); err != nil {
+                return fmt.Errorf("open operator: %w", err)
+        }
+        defer func() {
+                // ctx is usually already cancelled by the time we get here — that is
+                // why the task stopped. Close still needs a live window to flush, so
+                // give it one that keeps ctx's values but drops the cancellation.
+                closeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
+                defer cancel()
+
+                if cerr := op.Close(closeCtx); cerr != nil && err == nil {
+                        err = fmt.Errorf("close operator: %w", cerr)
+                }
+        }()
+
+        for {
+                record, ok := input.Receive(ctx)
+                if !ok {
+                        break
+                }
+
+                if err := op.Process(ctx, record, output); err != nil {
+                        return fmt.Errorf("process record: %w", err)
+                }
+        }
+
+        // Receive also returns !ok on cancellation, so report that as an error
+        // rather than a clean drain.
+        return ctx.Err()
+}
diff --git a/state/memory_backend.go b/state/memory_backend.go
new file mode 100644
index 0000000..6594657
--- /dev/null
+++ b/state/memory_backend.go
@@ -0,0 +1,57 @@
+package state
+
+import (
+        "context"
+        "flux/core"
+)
+
+// TODO: currently using Golang Map for go through the DAG pipline
+// This should be use index map index map[uint64]uint32  Hash(KeyGroup + StateName + Key) -> Offset
+// with 1GB of []byte allocation
+// This should be optimize with ZERO GC Scanning, Memory Alignment
+
+type MemoryStateBackend struct {
+        maxParallelism  int
+        currentKey      string
+        currentGroupKey int
+        currentNamespace any
+        // 4D structure StateName -> KeyGroup -> Key -> Namespace -> Value
+        states map[string]map[int]map[string]map[any]any
+}
+
+func NewMemoryStateBackend(maxParallelism int) *MemoryStateBackend {
+        return &MemoryStateBackend{
+                maxParallelism: maxParallelism,
+                states:         make(map[string]map[int]map[string]map[any]any),
+        }
+}
+
+func (ms *MemoryStateBackend) GetCurrentKey() string {
+        return ms.currentKey
+}
+
+func (ms *MemoryStateBackend) SetCurrentKey(key string) error {
+        ms.currentKey = key
+        ms.currentGroupKey = core.AssignToKeyGroup(key, ms.maxParallelism)
+        return nil
+}
+
+func (ms *MemoryStateBackend) GetValueState(ctx context.Context, key string) (core.ValueState[any], error) {
+        // Assign if state not existed
+
+        if _, ok := ms.states[key]; !ok {
+                ms.states[key] = make(map[int]map[string]map[any]any)
+        }
+
+        return &memoryStateValue{backend: ms, stateName: key}, nil
+}
+
+func (ms *MemoryStateBackend) GetCurrentNamespace() any{
+
+        return ms.currentNamespace
+}
+
+func (ms *MemoryStateBackend) SetCurrentNamespace(namespace any) error {
+        ms.currentNamespace = namespace
+        return nil
+}
diff --git a/state/memory_state.go b/state/memory_state.go
new file mode 100644
index 0000000..95c7318
--- /dev/null
+++ b/state/memory_state.go
@@ -0,0 +1,44 @@
+package state
+
+import (
+"context"
+)
+
+type memoryStateValue struct {
+        backend   *MemoryStateBackend
+        stateName string
+}
+
+func (m *memoryStateValue) Value(ctx context.Context) (any, error) {
+        kg := m.backend.currentGroupKey
+        key := m.backend.currentKey
+        ns := m.backend.currentNamespace
+        if kgMap, exist := m.backend.states[m.stateName][kg][key]; exist {
+                if v, hasKey := kgMap[ns]; hasKey {
+                        return v, nil
+                }
+        }
+
+        return nil, nil
+}
+
+func (m *memoryStateValue) Update(ctx context.Context, newVal any) (any, error) {
+        kg := m.backend.currentGroupKey
+        key := m.backend.currentKey
+        ns := m.backend.currentNamespace
+        if _, exists := m.backend.states[m.stateName][kg][key]; !exists {
+                m.backend.states[m.stateName][kg] = make(map[string]map[any]any)
+        }
+
+        m.backend.states[m.stateName][kg][key][ns] = newVal
+        return newVal, nil
+}
+
+func (m *memoryStateValue) Clear(ctx context.Context) error {
+        kg := m.backend.currentGroupKey
+        key := m.backend.currentKey
+        if kgMap, exists := m.backend.states[m.stateName][kg]; exists {
+                delete(kgMap, key)
+        }
+        return nil
+}
diff --git a/transport/channel.go b/transport/channel.go
new file mode 100644
index 0000000..56d5f2b
--- /dev/null
+++ b/transport/channel.go
@@ -0,0 +1,23 @@
+package transport
+
+import (
+        "context"
+
+        "flux/core"
+)
+
+type Receiver[T any] interface {
+        // Receive returns the next record, or ok=false once the stream is done or
+        // ctx is cancelled.
+        Receive(ctx context.Context) (core.StreamRecord[T], bool)
+        Close() error
+}
+
+// Emitter mirrors core.Emitter plus Close. It is declared separately because
+// core cannot import transport without an import cycle.
+type Emitter[T any] interface {
+        // Emit blocks until the record is handed off, or returns ctx.Err() if ctx
+        // is cancelled first.
+        Emit(ctx context.Context, record core.StreamRecord[T]) error
+        Close() error
+}
diff --git a/transport/local_channel.go b/transport/local_channel.go
new file mode 100644
index 0000000..2fac029
--- /dev/null
+++ b/transport/local_channel.go
@@ -0,0 +1,56 @@
+package transport
+
+import (
+        "context"
+
+        "flux/core"
+)
+
+type LocalReceiveChannel[T any] struct {
+        recordCh <-chan core.StreamRecord[T]
+}
+
+func NewLocalReceiveChannel[T any](ch <-chan core.StreamRecord[T]) *LocalReceiveChannel[T] {
+        return &LocalReceiveChannel[T]{
+                recordCh: ch,
+        }
+}
+
+func (c *LocalReceiveChannel[T]) Receive(ctx context.Context) (core.StreamRecord[T], bool) {
+        select {
+        case val, ok := <-c.recordCh:
+                return val, ok
+        case <-ctx.Done():
+                var zero core.StreamRecord[T]
+                return zero, false
+        }
+}
+
+func (c *LocalReceiveChannel[T]) Close() error {
+        // The emitter will close
+        return nil
+}
+
+type LocalEmitChannel[T any] struct {
+        recordCh chan<- core.StreamRecord[T]
+}
+
+func NewLocalEmitChannel[T any](ch chan<- core.StreamRecord[T]) *LocalEmitChannel[T] {
+        return &LocalEmitChannel[T]{
+                recordCh: ch,
+        }
+}
+
+func (l *LocalEmitChannel[T]) Emit(ctx context.Context, c core.StreamRecord[T]) error {
+        select {
+        case l.recordCh <- c:
+                return nil
+        case <-ctx.Done():
+                return ctx.Err()
+        }
+}
+
+func (l *LocalEmitChannel[T]) Close() error {
+        close(l.recordCh)
+        return nil
+}