feat: add on timer & watermarkmain

1 files changed, 34 insertions, 0 deletions

diff --git a/core/state_backend.go b/core/state_backend.go
index f8b83fb..729f0fe 100644
--- a/core/state_backend.go
+++ b/core/state_backend.go
@@ -3,11 +3,15 @@ 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 string) error
+	GetCurrentNamespace() string
 
+	// For User ===
 	// GetValueState will perform in disk
 	GetValueState(ctx context.Context, key string) (ValueState[any], error)
 }
@@ -16,6 +20,36 @@ 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)
 }