package state

import (
	"context"
	"encoding/binary"
	"fmt"
	"os"
	"path/filepath"
	"sync"

	"flux/core"

	"github.com/cockroachdb/pebble/v2"
)

type PebbleStateBackend struct {
	maxParallelism   int
	currentKey       string
	currentGroupKey  int
	currentNamespace any
	db               *pebble.DB

	// for support statename storage optimize
	// PebbleDB dont have column family as RockDB does
	// manually convert name <-> id
	mu       sync.Mutex
	stateMap map[string]uint8
	// nextStateID prevent the hash collision
	nextStateID uint8
}

func NewPebbleStateBackend(dir string, maxParalleism int) (*PebbleStateBackend, error) {
	db, err := pebble.Open(dir, &pebble.Options{})
	if err != nil {
		return nil, fmt.Errorf("failed to open pebbledb %w", err)
	}
	return &PebbleStateBackend{
		maxParallelism:   maxParalleism,
		db:               db,
		stateMap:         make(map[string]uint8),
		nextStateID:      1,
		currentNamespace: &core.VoidNamespace{},
	}, nil
}

func (p *PebbleStateBackend) SetCurrentKey(key string) error {
	p.currentKey = key
	p.currentGroupKey = core.AssignToKeyGroup(key, p.maxParallelism)
	return nil
}

func (p *PebbleStateBackend) GetCurrentKey() string {
	return p.currentKey
}

func (p *PebbleStateBackend) SetCurrentNamespace(namespace any) error {
	p.currentNamespace = namespace
	return nil
}

func (p *PebbleStateBackend) GetCurrentNamespace() any {
	return p.currentNamespace
}

// 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 &pebbleStateValue{
		backend: p,
		stateID: stateID,
	}, nil
}

func (p *PebbleStateBackend) getStateID(stateName string) uint8 {
	p.mu.Lock()
	defer p.mu.Unlock()

	if _, ok := p.stateMap[stateName]; !ok {

		id := p.nextStateID
		p.stateMap[stateName] = id
		p.nextStateID++
	}

	return p.stateMap[stateName]
}

func (p *PebbleStateBackend) calculateNamespaceSize() int {
	nsLen := 0
	switch p.currentNamespace.(type) {
	case core.TimeWindow:
		nsLen = 16
	case string:
		nsLen = len(p.currentNamespace.(string))
	default:
		// void namespace
		nsLen = 0
	}

	return nsLen
}

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)

	kl := len(p.currentKey)
	b := make([]byte, 0, p.calculateNamespaceSize()+5+kl)

	// add keygroup
	b = binary.BigEndian.AppendUint16(b, uint16(p.currentGroupKey))

	// add state id
	b = append(b, stateID)

	// add key
	b = binary.BigEndian.AppendUint16(b, uint16(kl))
	b = append(b, p.currentKey...)

	// add window / namespace

	switch ns := p.currentNamespace.(type) {
	case core.TimeWindow:
		b = binary.BigEndian.AppendUint64(b, uint64(ns.Start))
		b = binary.BigEndian.AppendUint64(b, uint64(ns.End))
	case string:
		b = append(b, ns...)
	default:
	}
	return b
}

func (p *PebbleStateBackend) Close() error {
	return p.db.Close()
}

// Snapshot atomic
func (p *PebbleStateBackend) Snapshot(checkpointID uint64, targetDir string) error {
	// get parent dir
	parentDir := filepath.Dir(targetDir)

	if err := os.MkdirAll(parentDir, 0o755); err != nil {
		return err
	}

	// create temp dir
	tmpDir := targetDir + ".tmp"
	_ = os.RemoveAll(tmpDir)

	// rollback
	success := false
	defer func() {
		if !success {
			_ = os.RemoveAll(tmpDir)
		}
	}()

	if err := p.db.Checkpoint(tmpDir); err != nil {
		return fmt.Errorf("pebble checkpoint failed for ID %d: %w", checkpointID, err)
	}

	_ = os.RemoveAll(targetDir)

	if err := os.Rename(tmpDir, targetDir); err != nil {
		return fmt.Errorf("failed to commit checkpoint atomically from %s to %s: %w", tmpDir, targetDir, err)
	}

	// fsync the parent dir
	if parentFile, err := os.Open(parentDir); err == nil {
		_ = parentFile.Sync()
		_ = parentFile.Close()
	}

	success = true 
	return nil
}