From e626cee8f7de0dd3ffca4c7297e59a87f790cc02 Mon Sep 17 00:00:00 2001
From: sangtran <tranquangsang12.7@gmail.com>
Date: Sat, 5 Sep 2026 17:06:30 +0700
Subject: feat: add on pebble state backend and using sync Pool to reuse buffer
 []byte

---
 state/memory_backend.go |  8 ++++----
 state/pebble_backend.go | 30 +++++++++++++++------------
 state/pebble_state.go   | 54 +++++++++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 75 insertions(+), 17 deletions(-)
 create mode 100644 state/pebble_state.go

(limited to 'state')

diff --git a/state/memory_backend.go b/state/memory_backend.go
index dda577e..cce8194 100644
--- a/state/memory_backend.go
+++ b/state/memory_backend.go
@@ -39,15 +39,15 @@ func (ms *MemoryStateBackend) SetCurrentKey(key string) error {
 	return nil
 }
 
-func (ms *MemoryStateBackend) GetValueState(ctx context.Context, statename string) (core.ValueState[any], error) {
+func (ms *MemoryStateBackend) GetValueState(ctx context.Context, stateName string) (core.ValueState[any], error) {
 	// Assign if state not existed
 	
-	if _, ok := ms.states[statename]; !ok {
+	if _, ok := ms.states[stateName]; !ok {
 		// create state name map if not existed 
-		ms.states[statename] = make(map[int]map[string]map[any]any)
+		ms.states[stateName] = make(map[int]map[string]map[any]any)
 	}
 
-	return &memoryStateValue{backend: ms, stateName: statename}, nil
+	return &memoryStateValue{backend: ms, stateName: stateName}, nil
 }
 
 func (ms *MemoryStateBackend) GetCurrentNamespace() any {
diff --git a/state/pebble_backend.go b/state/pebble_backend.go
index e428988..fbb8d68 100644
--- a/state/pebble_backend.go
+++ b/state/pebble_backend.go
@@ -33,10 +33,11 @@ func NewPebbleStateBackend(dir string, maxParalleism int) (*PebbleStateBackend,
 		return nil, fmt.Errorf("failed to open pebbledb %w", err)
 	}
 	return &PebbleStateBackend{
-		maxParallelism: maxParalleism,
-		db:             db,
-		stateMap:       make(map[string]uint8),
-		nextStateID:    1,
+		maxParallelism:   maxParalleism,
+		db:               db,
+		stateMap:         make(map[string]uint8),
+		nextStateID:      1,
+		currentNamespace: &core.VoidNamespace{},
 	}, nil
 }
 
@@ -60,23 +61,27 @@ func (p *PebbleStateBackend) GetCurrentNamespace() any {
 }
 
 // GetValueState(ctx context.Context, key string) (ValueState[any], error)
-func (p *PebbleStateBackend) GetValueState(ctx context.Context, statename string) (core.ValueState[any], error) {
-	stateID := p.getStateID(statename)
-	return nil, nil
+func (p *PebbleStateBackend) GetValueState(ctx context.Context, stateName string) (core.ValueState[any], error) {
+	stateID := p.getStateID(stateName)
+
+	return &pebbleStateValue{
+		backend: p,
+		stateID: stateID,
+	}, nil
 }
 
-func (p *PebbleStateBackend) getStateID(statename string) uint8 {
+func (p *PebbleStateBackend) getStateID(stateName string) uint8 {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 
-	if _, ok := p.stateMap[statename]; !ok {
+	if _, ok := p.stateMap[stateName]; !ok {
 
 		id := p.nextStateID
-		p.stateMap[statename] = id
+		p.stateMap[stateName] = id
 		p.nextStateID++
 	}
 
-	return p.stateMap[statename]
+	return p.stateMap[stateName]
 }
 
 func (p *PebbleStateBackend) calculateNamespaceSize() int {
@@ -94,7 +99,7 @@ func (p *PebbleStateBackend) calculateNamespaceSize() int {
 	return nsLen
 }
 
-func (p *PebbleStateBackend) deserialize(statename string) []byte {
+func (p *PebbleStateBackend) deserialize(stateID uint8) []byte {
 	// this will store in SSTable, order is important, keygroup MUST be first for scaling
 	// 2B(KeyGroup) + 1B(StateID) + 2B(KeyLen) + len(Key) + 16B(Window)
 
@@ -105,7 +110,6 @@ func (p *PebbleStateBackend) deserialize(statename string) []byte {
 	b = binary.BigEndian.AppendUint16(b, uint16(p.currentGroupKey))
 
 	// add state id
-	stateID := p.getStateID(statename)
 	b = append(b, stateID)
 
 	// add key
diff --git a/state/pebble_state.go b/state/pebble_state.go
new file mode 100644
index 0000000..3eeaa05
--- /dev/null
+++ b/state/pebble_state.go
@@ -0,0 +1,54 @@
+package state
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"flux/core"
+
+	"github.com/cockroachdb/pebble/v2"
+)
+
+type pebbleStateValue struct {
+	backend *PebbleStateBackend
+	stateID uint8
+}
+
+func (p *pebbleStateValue) Update(ctx context.Context, newVal any) (any, error) {
+	key := p.backend.deserialize(p.stateID)
+	// convert newVal to []byte raw
+	v, err := json.Marshal(newVal)
+	if err != nil {
+		return nil, err
+	}
+
+	err = p.backend.db.Set(key, v, pebble.NoSync)
+	if err != nil {
+		return nil, err
+	}
+
+	return newVal, nil
+}
+
+func (p *pebbleStateValue) Value(ctx context.Context) (any, error) {
+	key := p.backend.deserialize(p.stateID)
+	v, closer, err := p.backend.db.Get(key)
+	if err != nil {
+		if errors.Is(err, pebble.ErrNotFound) {
+			return nil, nil 
+		}
+		return nil, err
+	}
+
+	defer closer.Close()
+
+	res := core.GetBuffer(len(v))
+	copy(res, v)
+	return v, nil
+}
+
+func (p *pebbleStateValue) Clear(ctx context.Context) error {
+	key := p.backend.deserialize(p.stateID)
+
+	return p.backend.db.Delete(key, pebble.NoSync)
+}
-- 
cgit v1.3.1