capacitor.go

package capacitor

import (
	"context"
	"crypto/sha256"
	"crypto/tls"
	"errors"
	"fmt"
	"log"
	"math/bits"
	"net"
	"os"
	"reflect"
	"runtime"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/hashicorp/memberlist"
	"github.com/vmihailenco/msgpack/v5"
)

const NoExpiry = time.Duration(-1)

var ErrKeyNotFound = errors.New("capacitor: key not found")

// Logger is a generic structured logging interface.
type Logger interface {
	Debug(msg string, args ...any)
	Info(msg string, args ...any)
	Warn(msg string, args ...any)
	Error(msg string, args ...any)
}

type noopLogger struct{}

func (n noopLogger) Debug(msg string, args ...any) {}
func (n noopLogger) Info(msg string, args ...any)  {}
func (n noopLogger) Warn(msg string, args ...any)  {}
func (n noopLogger) Error(msg string, args ...any) {}

// Config defines the configuration parameters for a Capacitor node instance.
type Config struct {
	// NodeID is the unique identifier for this node in the cluster. If left empty,
	// a hostname-based identifier will be generated automatically.
	NodeID string

	// BindAddr is the network address for the gossip memberlist to bind to.
	BindAddr string

	// BindPort is the port number utilized for gossip memberlist communications.
	BindPort int

	// StreamPort is the port number utilized for replication TCP streams.
	StreamPort int

	// AdvertiseAddr is the IP address advertised to other nodes for establishing
	// replication TCP streams.
	AdvertiseAddr string

	// Peers is the initial list of bootstrap addresses ("IP:port") of active cluster nodes.
	Peers []string

	// DataPath is the local directory path where BadgerDB files are persistently stored.
	DataPath string

	// LogSize is the capacity (maximum number of entries) of the in-memory circular Delta Log.
	LogSize uint64

	// TLSConfig is the optional configuration used to secure node-to-node replication streams using mTLS.
	TLSConfig *tls.Config

	// AuthToken is a shared secret token used to authenticate gossip join requests and TCP streams.
	AuthToken string

	// Logger is the structured logging engine injected into the Capacitor instance.
	Logger Logger

	// DisableMetrics disables internal metrics latency tracking for maximum read/write performance.
	DisableMetrics bool
}

// Capacitor represents an active-active, local-first replicated caching node.
// It manages local key-value storage, cluster membership discovery, logical clocks,
// and replication orchestration.
type Capacitor struct {
	nodeID    string
	store     *store
	hlc       *HLC
	log       *DeltaLog
	server    *StreamServer
	client    *StreamClient
	ml              *memberlist.Memberlist
	metrics         *MetricsTracker
	authToken       string
	hashedAuthToken string
	logger          Logger
	ctx             context.Context
	cancel          context.CancelFunc

	// Replicator state

	peerSeqs       sync.Map // map[string]uint64 (Last replicated Seq)
	peerMutex      sync.Mutex
	peerStops      map[string]peerReplicator
	streamAddr     string
	stop           chan struct{}
	disableMetrics bool
	pubSub         *pubSubRegistry
	subCounter     uint64
	replicatorWG   sync.WaitGroup
	closing        uint32 // Atomic flag indicating that the node is closing
}

type peerReplicator struct {
	stop chan struct{}
	done chan struct{}
}

// New initializes, configures, and starts a local Capacitor node.
// This constructs the storage, starts the replication stream listener, and registers
// the node with the gossip cluster.
func New(cfg Config) (*Capacitor, error) {
	if cfg.NodeID == "" {
		hostname, _ := os.Hostname()
		cfg.NodeID = fmt.Sprintf("%s-%d", hostname, time.Now().Unix())
	}
	if cfg.LogSize == 0 {
		cfg.LogSize = 1_000_000
	}
	if cfg.Logger == nil {
		cfg.Logger = noopLogger{}
	}
	if cfg.BindPort < 0 || cfg.BindPort > 65535 {
		return nil, errors.New("BindPort must be between 0 and 65535")
	}
	if cfg.StreamPort < 0 || cfg.StreamPort > 65535 {
		return nil, errors.New("StreamPort must be between 0 and 65535")
	}

	bindPort := cfg.BindPort
	if bindPort == 0 {
		bindPort = 7946
	}
	if cfg.StreamPort != 0 && cfg.StreamPort == bindPort {
		return nil, errors.New("StreamPort and BindPort cannot be the same")
	}

	if cfg.AuthToken == "" {
		cfg.Logger.Warn("AuthToken is empty. Replication stream addresses and node metadata will be transmitted in plaintext via insecure gossip protocols. Do not use this configuration in production.")
	}
	if cfg.DataPath == "" {
		cfg.Logger.Info("DataPath is empty. Running BadgerDB in-memory mode. Data will not persist across restarts.")
	}

	metrics := NewMetricsTracker()
	s, err := newStore(cfg.DataPath, metrics, cfg.Logger)
	if err != nil {
		metrics.Stop()
		return nil, err
	}

	ctx, cancel := context.WithCancel(context.Background())
	cp := &Capacitor{
		nodeID:         cfg.NodeID,
		store:          s,
		hlc:            NewHLC(),
		log:            NewDeltaLog(cfg.LogSize),
		client:         NewStreamClient(cfg.TLSConfig, cfg.AuthToken, metrics),
		metrics:        metrics,
		stop:           make(chan struct{}),
		peerStops:      make(map[string]peerReplicator),
		authToken:       cfg.AuthToken,
		hashedAuthToken: hashToken(cfg.AuthToken),
		logger:          cfg.Logger,
		ctx:             ctx,
		cancel:          cancel,
		disableMetrics:  cfg.DisableMetrics,
		pubSub:          newPubSubRegistry(),
	}

	srv, err := NewStreamServer(cp, fmt.Sprintf("0.0.0.0:%d", cfg.StreamPort), cfg.TLSConfig)
	if err != nil {
		return nil, err
	}
	cp.server = srv
	go srv.Start()

	// Memberlist config for discovery (initialized early so we can read BindAddr)
	mlCfg := memberlist.DefaultLocalConfig()
	mlCfg.Name = cfg.NodeID
	if cfg.BindAddr != "" {
		mlCfg.BindAddr = cfg.BindAddr
	}
	mlCfg.BindPort = cfg.BindPort

	// Secure Gossip layer with AuthToken
	if cfg.AuthToken != "" {
		hash := sha256.Sum256([]byte(cfg.AuthToken))
		mlCfg.SecretKey = hash[:]
	}

	// Bridge memberlist logging
	mlCfg.Logger = log.New(&mlLoggerBridge{cp.logger}, "", 0)

	// Determine the actual port bound for the stream server
	_, portStr, _ := net.SplitHostPort(srv.listener.Addr().String())

	// Determine the address to advertise for TCP streams
	advertiseAddr := cfg.AdvertiseAddr
	if advertiseAddr == "" {
		if mlCfg.BindAddr == "" || mlCfg.BindAddr == "0.0.0.0" || mlCfg.BindAddr == "127.0.0.1" || mlCfg.BindAddr == "localhost" {
			if isRunningTests() {
				advertiseAddr = "127.0.0.1"
			} else {
				advertiseAddr = getLocalIP()
			}
		} else {
			advertiseAddr = mlCfg.BindAddr
		}
	}
	cp.streamAddr = net.JoinHostPort(advertiseAddr, portStr)

	events := &clusterEvents{cp: cp}
	mlCfg.Delegate = events
	mlCfg.Events = events

	ml, err := memberlist.Create(mlCfg)
	if err != nil {
		return nil, err
	}
	cp.ml = ml

	if len(cfg.Peers) > 0 {
		_, _ = ml.Join(cfg.Peers)
	}

	return cp, nil
}

type clusterEvents struct {
	cp *Capacitor
}

type mlLoggerBridge struct {
	l Logger
}

func (m *mlLoggerBridge) Write(p []byte) (n int, err error) {
	// Memberlist logs are usually prefixed with [DEBUG], [ERR], etc.
	msg := string(p)
	switch {
	case strings.Contains(msg, "[DEBUG]"):
		m.l.Debug(strings.TrimSpace(msg))
	case strings.Contains(msg, "[ERR]"):
		m.l.Error(strings.TrimSpace(msg))
	case strings.Contains(msg, "[WARN]"):
		m.l.Warn(strings.TrimSpace(msg))
	default:
		m.l.Info(strings.TrimSpace(msg))
	}
	return len(p), nil
}

func (n *clusterEvents) NodeMeta(limit int) []byte {
	// Simple handshake: share the address
	// In a more complex setup, we could encode a map of LastSeenSeqs here.
	// For now, let's just stick to the address and let replicators start from store state.
	return []byte(n.cp.streamAddr)
}
func (n *clusterEvents) NotifyMsg([]byte)                           {}
func (n *clusterEvents) GetBroadcasts(overhead, limit int) [][]byte { return nil }
func (n *clusterEvents) LocalState(join bool) []byte                { return nil }
func (n *clusterEvents) MergeRemoteState(buf []byte, join bool)     {}

func (n *clusterEvents) NotifyJoin(node *memberlist.Node) {
	if node.Name == n.cp.nodeID {
		return
	}
	n.cp.startPeerReplicator(node.Name, string(node.Meta), false)
}

func (n *clusterEvents) NotifyLeave(node *memberlist.Node) {
	if n.cp.client != nil {
		n.cp.client.CloseConn(node.Name)
	}
	n.cp.stopPeerReplicator(node.Name)
}

func (n *clusterEvents) NotifyUpdate(node *memberlist.Node) {}

func (f *Capacitor) startPeerReplicator(nodeID, addr string, force bool) {
	if addr == "" {
		return
	}

	if atomic.LoadUint32(&f.closing) == 1 {
		return
	}

	f.peerMutex.Lock()
	defer f.peerMutex.Unlock()

	// If there's an existing replicator, stop it and wait for it to exit
	if stopVal, ok := f.peerStops[nodeID]; ok {
		if !force {
			return
		}
		close(stopVal.stop)
		<-stopVal.done
		delete(f.peerStops, nodeID)
	}

	stop := make(chan struct{})
	done := make(chan struct{})
	pr := peerReplicator{stop: stop, done: done}
	f.peerStops[nodeID] = pr

	f.replicatorWG.Add(1)
	go func() {
		defer f.replicatorWG.Done()
		defer close(done)
		f.peerReplicatorLoop(nodeID, addr, stop)
	}()
}

func (f *Capacitor) stopPeerReplicator(nodeID string) {
	f.peerMutex.Lock()
	if stopVal, ok := f.peerStops[nodeID]; ok {
		close(stopVal.stop)
		<-stopVal.done
		delete(f.peerStops, nodeID)
	}
	f.peerMutex.Unlock()

	f.log.mu.Lock()
	close(f.log.notifyCh)
	f.log.notifyCh = make(chan struct{})
	f.log.mu.Unlock()
	f.peerSeqs.Delete(nodeID)
}

func (f *Capacitor) peerReplicatorLoop(nodeID, addr string, stop chan struct{}) {
	defer func() {
		if r := recover(); r != nil {
			f.logger.Error("panic in peerReplicatorLoop", "error", r, "nodeID", nodeID, "addr", addr)
		}
	}()
	for {
		var ch chan struct{}
		var lastSeq uint64

		f.log.mu.Lock()
		for {
			select {
			case <-stop:
				f.log.mu.Unlock()
				return
			case <-f.ctx.Done():
				f.log.mu.Unlock()
				return
			default:
			}

			lastSeqVal, ok := f.peerSeqs.Load(nodeID)
			if ok && lastSeqVal != nil {
				if seq, seqOk := lastSeqVal.(uint64); seqOk {
					lastSeq = seq
				} else {
					lastSeq = f.store.getPeerSeq(nodeID)
				}
			} else {
				lastSeq = f.store.getPeerSeq(nodeID)
			}

			// head is the next sequence ID to be assigned.
			// Entries exist up to head-1.
			// If lastSeq is the last one we sent, we want to wait until head > lastSeq + 1.
			if f.log.head > lastSeq+1 {
				break
			}

			ch = f.log.notifyCh
			f.log.mu.Unlock()

			select {
			case <-stop:
				return
			case <-f.ctx.Done():
				return
			case <-ch:
				f.log.mu.Lock()
			}
		}
		f.log.mu.Unlock()

		// 2. Perform replication
		success := f.replicateToPeer(f.ctx, nodeID, addr)
		if !success {
			select {
			case <-stop:
				return
			case <-f.ctx.Done():
				return
			case <-time.After(100 * time.Millisecond):
			}
		}
	}
}

func (f *Capacitor) replicateToPeer(ctx context.Context, nodeID, addr string) bool {
	lastSeqVal, _ := f.peerSeqs.LoadOrStore(nodeID, f.store.getPeerSeq(nodeID))
	var lastSeq uint64
	if lastSeqVal != nil {
		if seq, seqOk := lastSeqVal.(uint64); seqOk {
			lastSeq = seq
		} else {
			lastSeq = f.store.getPeerSeq(nodeID)
		}
	} else {
		lastSeq = f.store.getPeerSeq(nodeID)
	}

	f.log.mu.RLock()
	earliest := f.log.earliestSeq
	f.log.mu.RUnlock()

	if lastSeq+1 < earliest {
		f.logger.Warn("Replication gap detected, performing full state bootstrap catch-up", "peer", nodeID, "lastSeq", lastSeq, "earliestSeq", earliest)

		snapEntries := f.store.getSnapshotEntries()

		const batchSize = 1000
		for idx := 0; idx < len(snapEntries); idx += batchSize {
			end := idx + batchSize
			if end > len(snapEntries) {
				end = len(snapEntries)
			}

			var rawSnap [][]byte
			for _, entry := range snapEntries[idx:end] {
				bin, _ := entry.MarshalMsg(nil)
				rawSnap = append(rawSnap, bin)
			}

			batch := Batch{
				FromNode: f.nodeID,
				Entries:  rawSnap,
			}

			repCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
			lastSeenRemote := f.store.getPeerSeq(nodeID)
			err := f.client.SendBatch(repCtx, nodeID, addr, batch, lastSeenRemote)
			cancel()

			if err != nil {
				f.logger.Error("Full state sync batch send failed", "peer", nodeID, "error", err)
				atomic.AddUint64(&f.metrics.ReplicationFailures, 1)
				return false
			}
		}

		f.peerSeqs.Store(nodeID, earliest-1)
		f.logger.Info("Full state bootstrap catch-up complete, resuming delta replication", "peer", nodeID, "newSeq", earliest-1)
		return true
	}

	// 1. Get new entries from log as raw binary slices
	// Hard-limit to 1000 entries OR 8MB per batch for backpressure
	rawEntries := f.log.GetEntriesRaw(lastSeq+1, 1000, 8*1024*1024)
	if len(rawEntries) == 0 {
		return true
	}
	defer f.log.PutEntriesRaw(rawEntries)

	batch := Batch{
		FromNode: f.nodeID,
		Entries:  rawEntries,
	}

	// 2. Stream to this specific peer
	// Pass our LastSeenSeq from them to initiate handshake
	lastSeenRemote := f.store.getPeerSeq(nodeID)

	// Create a sub-context with timeout for the replication batch
	repCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
	defer cancel()

	success := true
	if err := f.client.SendBatch(repCtx, nodeID, addr, batch, lastSeenRemote); err == nil {
		// Only advance on success
		f.peerSeqs.Store(nodeID, lastSeq+uint64(len(rawEntries)))
	} else {
		atomic.AddUint64(&f.metrics.ReplicationFailures, 1)
		success = false
	}
	return success
}

func (f *Capacitor) applyRemoteEntry(ctx context.Context, e LogEntry, senderNodeID string) {
	if senderNodeID != "" && e.Seq > 0 {
		lastSeq := f.store.getPeerSeq(senderNodeID)
		if int64(e.Seq-lastSeq) <= 0 {
			return
		}
	}

	// Track replication latency
	if e.BornAt > 0 {
		f.metrics.ReplicateLat.Record(time.Since(time.Unix(0, e.BornAt)))
	}

	// Update local HLC
	if _, err := f.hlc.Update(e.TS); err != nil {
		// Log clock smash and drop the entry to prevent poisoning
		return
	}

	switch e.Op {
	case MsgSet:
		var val any
		_ = msgpack.Unmarshal(e.Value, &val)
		f.store.set(e.Key, val, e.TS, time.Duration(e.TTL)*time.Millisecond)
	case MsgIncr:
		f.store.setNodeCount(e.Key, e.NodeID, e.Delta)
	case MsgMetric:
		// Metrics are eventually consistent across nodes
		f.store.incrementMetric(e.Key, e.NodeID, e.Delta)
	case MsgWindow:
		f.store.addWindowTimestamp(e.Key, e.TS.Physical)
	case MsgDelete:
		f.store.delete(e.Key, e.TS)
	case MsgSetAdd:
		f.store.setAdd(e.Key, string(e.Value), e.TS)
	case MsgSetRemove:
		_, _ = f.store.setRemove(e.Key, string(e.Value), e.TS, true)
	case MsgSortedSetAdd:
		f.store.sortedSetAdd(e.Key, string(e.Value), e.Delta, e.TS)
	case MsgSortedSetIncrement:
		_, _ = f.store.sortedSetIncrementBy(e.Key, string(e.Value), e.Delta, e.TS)
	case MsgSortedSetRemove:
		_, _ = f.store.sortedSetRemove(e.Key, string(e.Value), e.TS, true)
	case MsgMapSet:
		var p MapPayload
		if err := msgpack.Unmarshal(e.Value, &p); err == nil {
			var ttl time.Duration
			if p.ExpiresAt > 0 {
				ttl = time.Duration(p.ExpiresAt-time.Now().UnixNano()) * time.Nanosecond
				if ttl < 0 {
					ttl = 1 * time.Nanosecond
				}
			}
			f.store.mapSet(e.Key, p.Field, p.Value, ttl, e.TS)
		}
	case MsgMapRemove:
		var p MapPayload
		if err := msgpack.Unmarshal(e.Value, &p); err == nil {
			f.store.mapRemove(e.Key, p.Field, e.TS)
		}
	case MsgNMapSet:
		var p NMapPayload
		if err := msgpack.Unmarshal(e.Value, &p); err == nil {
			var ttl time.Duration
			if p.ExpiresAt > 0 {
				ttl = time.Duration(p.ExpiresAt-time.Now().UnixNano()) * time.Nanosecond
				if ttl < 0 {
					ttl = 1 * time.Nanosecond
				}
			}
			_, _, _, _ = f.store.nmapSet(e.Key, p.Field, p.Value, ttl, e.TS, true)
		}
	case MsgNMapRemove:
		var p NMapPayload
		if err := msgpack.Unmarshal(e.Value, &p); err == nil {
			_, _, _, _, _ = f.store.nmapRemove(e.Key, p.Field, e.TS, true)
		}
	case MsgListInsert:
		var p ListPayload
		if err := msgpack.Unmarshal(e.Value, &p); err == nil {
			_, _ = f.store.listInsert(e.Key, p.ID, p.ParentID, p.Value, e.TS)
		}
	case MsgListDelete:
		var p ListPayload
		if err := msgpack.Unmarshal(e.Value, &p); err == nil {
			_, _ = f.store.listDelete(e.Key, p.ID, e.TS)
		}
	case MsgHLLAdd:
		var p HLLPayload
		if err := msgpack.Unmarshal(e.Value, &p); err == nil {
			_, _ = f.store.hllAdd(e.Key, p.Index, p.Value, e.TS)
		}
	case MsgBloomAdd:
		var p BloomPayload
		if err := msgpack.Unmarshal(e.Value, &p); err == nil {
			_, _ = f.store.bloomAdd(e.Key, p.Indices, e.TS)
		}
	case MsgCMSIncrement:
		var p CMSPayload
		if err := msgpack.Unmarshal(e.Value, &p); err == nil {
			_ = f.store.cmsIncrement(e.Key, p.Element, p.Count, e.TS)
		}
	case MsgPubSubPublish:
		var p PubSubPayload
		if err := msgpack.Unmarshal(e.Value, &p); err == nil {
			f.pubSub.publish(p.Topic, PubSubMessage{Topic: p.Topic, Payload: p.Payload})
		}
	}

	// Record persistent sequence for catch-up after restart
	if senderNodeID != "" && e.Seq > 0 {
		f.store.updatePeerSeq(senderNodeID, e.Seq)
	}
}

// Set writes a key-value pair to the local store and appends it to the replication
// log to propagate it to other nodes in the cluster. If a positive TTL is specified,
// the key-value pair will automatically expire.
func (f *Capacitor) Set(ctx context.Context, key string, value any, ttl time.Duration) error {
	var start time.Time
	if !f.disableMetrics {
		start = time.Now()
	}

	// 1. Serialize Once for Log/Replication
	binVal, err := msgpack.Marshal(value)
	if err != nil {
		return err
	}

	ts := f.hlc.Now()

	// 2. Local Write
	if err := f.store.set(key, value, ts, ttl); err != nil {
		return err
	}

	// 3. Log Append
	f.log.Append(LogEntry{
		TS:     ts,
		BornAt: time.Now().UnixNano(),
		Op:     MsgSet,
		Key:    key,
		Value:  binVal,
		TTL:    int64(ttl.Milliseconds()),
	})
	if !f.disableMetrics {
		f.metrics.SetLat.Record(time.Since(start))
	}
	return nil
}

// GetScan retrieves a key's value and unmarshals it into the destination pointer dest
// (similar to json.Unmarshal or database rows.Scan).
func (f *Capacitor) GetScan(ctx context.Context, key string, dest any) error {
	var start time.Time
	if !f.disableMetrics {
		start = time.Now()
	}

	err := f.getScanInternal(ctx, key, dest)

	if !f.disableMetrics {
		f.metrics.GetLat.Record(time.Since(start))
	}
	return err
}

func (f *Capacitor) getScanInternal(ctx context.Context, key string, dest any) error {
	rawVal, err := f.store.get(key)
	if err != nil {
		return err
	}
	if rawVal == nil {
		return fmt.Errorf("capacitor: key %s not found", key)
	}

	// 1. Reflection-based Direct Assignment Fast Path (In-Memory Hot Path Bypass)
	destVal := reflect.ValueOf(dest)
	if destVal.Kind() == reflect.Ptr {
		elemVal := destVal.Elem()
		if elemVal.Type() == reflect.TypeOf(rawVal) {
			elemVal.Set(reflect.ValueOf(rawVal))
			return nil
		}
	}

	// Direct assignment for simple types
	switch d := dest.(type) {
	case *string:
		switch v := rawVal.(type) {
		case string:
			*d = v
		case []byte:
			*d = string(v)
		default:
			b, _ := msgpack.Marshal(v)
			*d = string(b)
		}
		return nil
	case *[]byte:
		switch v := rawVal.(type) {
		case []byte:
			*d = v
		case string:
			*d = []byte(v)
		default:
			*d, _ = msgpack.Marshal(v)
		}
		return nil
	}

	// Fallback to MsgPack unmarshaling
	switch v := rawVal.(type) {
	case []byte:
		return msgpack.Unmarshal(v, dest)
	case string:
		return msgpack.Unmarshal([]byte(v), dest)
	default:
		b, err := msgpack.Marshal(v)
		if err != nil {
			return err
		}
		return msgpack.Unmarshal(b, dest)
	}
}

// Get retrieves a key-value pair's serialized value from the local cache database.
// Returns an empty string and nil error if the key is not found or has expired.
func (f *Capacitor) Get(ctx context.Context, key string) (string, error) {
	var start time.Time
	if !f.disableMetrics {
		start = time.Now()
	}

	res, err := f.getInternal(ctx, key)

	if !f.disableMetrics {
		f.metrics.GetLat.Record(time.Since(start))
	}
	return res, err
}

func (f *Capacitor) getInternal(ctx context.Context, key string) (string, error) {
	val, err := f.store.get(key)
	if err != nil {
		return "", err
	}
	if val == nil {
		return "", ErrKeyNotFound
	}

	res := ""
	switch v := val.(type) {
	case string:
		res = v
	case []byte:
		res = string(v)
	default:
		b, _ := msgpack.Marshal(v)
		res = string(b)
	}
	return res, nil
}

// Exists checks if a key exists in the cache and has not expired.
func (f *Capacitor) Exists(ctx context.Context, key string) (bool, error) {
	var start time.Time
	if !f.disableMetrics {
		start = time.Now()
	}

	exists, err := f.store.exists(key)

	if !f.disableMetrics {
		f.metrics.GetLat.Record(time.Since(start))
	}
	return exists, err
}

// Delete removes a key-value pair from the local store and replicates the deletion tombstone to the cluster.
func (f *Capacitor) Delete(ctx context.Context, key string) error {
	var start time.Time
	if !f.disableMetrics {
		start = time.Now()
	}
	ts := f.hlc.Now()

	// 1. Local Write
	if err := f.store.delete(key, ts); err != nil {
		return err
	}

	// 2. Log Append
	f.log.Append(LogEntry{
		TS:     ts,
		BornAt: time.Now().UnixNano(),
		Op:     MsgDelete,
		Key:    key,
	})

	if !f.disableMetrics {
		f.metrics.SetLat.Record(time.Since(start))
	}
	return nil
}

// IncrementBy increments a distributed PN-Counter key by the specified delta.
// It tracks counts per-node to construct CRDT conflict-free convergence.
func (f *Capacitor) IncrementBy(ctx context.Context, key string, delta int64) (int64, error) {
	var start time.Time
	if !f.disableMetrics {
		start = time.Now()
	}
	val, err := f.store.increment(key, f.nodeID, float64(delta))
	if err != nil {
		return 0, err
	}

	nodeVal, _ := f.store.getNodeCount(key, f.nodeID)
	f.log.Append(LogEntry{
		TS:     f.hlc.Now(),
		BornAt: time.Now().UnixNano(),
		Op:     MsgIncr,
		Key:    key,
		Delta:  nodeVal,
		NodeID: f.nodeID,
	})
	if !f.disableMetrics {
		f.metrics.IncrLat.Record(time.Since(start))
	}
	return int64(val), nil
}

// GetMetrics returns a snapshot summary of all built-in metrics (latencies, counts).
func (f *Capacitor) GetMetrics() []Summary {
	summaries := f.metrics.GetSummary()

	// Append Log Overflows
	summaries = append(summaries, Summary{
		Metric: "DeltaLog Overflows",
		Count:  int64(atomic.LoadUint64(&f.log.Overflows)),
	})

	// Append Clock Smashes
	f.hlc.mu.Lock()
	clockSmashes := f.hlc.ClockSmashes
	f.hlc.mu.Unlock()
	summaries = append(summaries, Summary{
		Metric: "Clock Smash Events",
		Count:  int64(clockSmashes),
	})

	// Append GC Runs and Tombstones Evicted
	summaries = append(summaries, Summary{
		Metric: "GC Runs",
		Count:  int64(atomic.LoadUint64(&f.store.gcRuns)),
	})
	summaries = append(summaries, Summary{
		Metric: "Tombstones Evicted",
		Count:  int64(atomic.LoadUint64(&f.store.tombstonesEvicted)),
	})

	return summaries
}

// GossipAddr returns the actual local address and port bound by the gossip layer.
func (f *Capacitor) GossipAddr() string {
	if f.ml == nil {
		return ""
	}
	return fmt.Sprintf("127.0.0.1:%d", f.ml.LocalNode().Port)
}

// Increment increments a distributed PN-Counter key by 1.
func (f *Capacitor) Increment(ctx context.Context, key string) (int64, error) {
	return f.IncrementBy(ctx, key, 1)
}

// GetCount retrieves the converged aggregate sum of a distributed counter across all nodes.
func (f *Capacitor) GetCount(ctx context.Context, key string) (int64, error) {
	val, err := f.store.getAggregateCount(key)
	return int64(val), err
}

// IncrementParallel performs concurrent increment calls for multiple counter keys.
func (f *Capacitor) IncrementParallel(ctx context.Context, keys []string) (map[string]int64, error) {
	if len(keys) == 0 {
		return make(map[string]int64), nil
	}
	if len(keys) == 1 {
		val, err := f.Increment(ctx, keys[0])
		if err != nil {
			return nil, err
		}
		return map[string]int64{keys[0]: val}, nil
	}

	res := make(map[string]int64, len(keys))
	var mu sync.Mutex
	var wg sync.WaitGroup
	var firstErr error
	var errOnce sync.Once

	for _, k := range keys {
		wg.Add(1)
		go func(key string) {
			defer wg.Done()
			val, err := f.Increment(ctx, key)
			if err != nil {
				errOnce.Do(func() { firstErr = err })
				return
			}
			mu.Lock()
			res[key] = val
			mu.Unlock()
		}(k)
	}

	wg.Wait()
	if firstErr != nil {
		return nil, firstErr
	}
	return res, nil
}

// IncrementMetric records a floating-point update to a distributed aggregate metric
// (tracking both hit frequency and aggregated sums).
func (f *Capacitor) IncrementMetric(ctx context.Context, key string, delta float64) (Metric, error) {
	m, err := f.store.incrementMetric(key, f.nodeID, delta)
	if err != nil {
		return Metric{}, err
	}

	f.log.Append(LogEntry{
		TS:     f.hlc.Now(),
		Op:     MsgMetric,
		Key:    key,
		Delta:  delta,
		NodeID: f.nodeID,
	})
	return m, nil
}

// GetMetric retrieves the aggregated Metric details (count, sum, average) for the specified key.
func (f *Capacitor) GetMetric(ctx context.Context, key string) (Metric, error) {
	return f.store.getAggregateMetric(key)
}

// IncrementMetricParallel updates multiple aggregate metrics concurrently.
func (f *Capacitor) IncrementMetricParallel(ctx context.Context, keys map[string]float64) (map[string]Metric, error) {
	if len(keys) == 0 {
		return make(map[string]Metric), nil
	}
	if len(keys) == 1 {
		for k, d := range keys {
			m, err := f.IncrementMetric(ctx, k, d)
			if err != nil {
				return nil, err
			}
			return map[string]Metric{k: m}, nil
		}
	}

	res := make(map[string]Metric, len(keys))
	var mu sync.Mutex
	var wg sync.WaitGroup
	var firstErr error
	var errOnce sync.Once

	for k, d := range keys {
		wg.Add(1)
		go func(key string, delta float64) {
			defer wg.Done()
			m, err := f.IncrementMetric(ctx, key, delta)
			if err != nil {
				errOnce.Do(func() { firstErr = err })
				return
			}
			mu.Lock()
			res[key] = m
			mu.Unlock()
		}(k, d)
	}

	wg.Wait()
	if firstErr != nil {
		return nil, firstErr
	}
	return res, nil
}

// IncrementSlidingWindow appends an event timestamp for rate limiting or hit tracking,
// and returns the count of active occurrences within the rolling window duration.
func (f *Capacitor) IncrementSlidingWindow(ctx context.Context, key string, window time.Duration) (int64, error) {
	ts := f.hlc.Now()
	count, err := f.store.incrementSlidingWindow(key, ts.Physical, window)
	if err != nil {
		return 0, err
	}

	f.log.Append(LogEntry{
		TS:     ts,
		Op:     MsgWindow,
		Key:    key,
		NodeID: f.nodeID,
	})
	return count, nil
}

// SetAdd adds a member to a distributed replicated set.
// The member can be of any comparable or serializable type (e.g. string, int, struct).
func (f *Capacitor) SetAdd(ctx context.Context, key string, member any) (bool, error) {
	var start time.Time
	if !f.disableMetrics {
		start = time.Now()
	}

	mKey, err := memberToKey(member)
	if err != nil {
		return false, err
	}

	isMem, err := f.store.setIsMember(key, mKey)
	if err != nil {
		return false, err
	}
	if isMem {
		if !f.disableMetrics {
			f.metrics.SetLat.Record(time.Since(start))
		}
		return false, nil
	}

	ts := f.hlc.Now()
	if err := f.store.setAdd(key, mKey, ts); err != nil {
		return false, err
	}

	f.log.Append(LogEntry{
		TS:     ts,
		BornAt: time.Now().UnixNano(),
		Op:     MsgSetAdd,
		Key:    key,
		Value:  []byte(mKey),
	})

	if !f.disableMetrics {
		f.metrics.SetLat.Record(time.Since(start))
	}
	return true, nil
}

// SetRemove removes a member from a distributed replicated set.
func (f *Capacitor) SetRemove(ctx context.Context, key string, member any) (bool, error) {
	var start time.Time
	if !f.disableMetrics {
		start = time.Now()
	}

	mKey, err := memberToKey(member)
	if err != nil {
		return false, err
	}

	ts := f.hlc.Now()
	removed, err := f.store.setRemove(key, mKey, ts, false)
	if err != nil {
		return false, err
	}

	if removed {
		f.log.Append(LogEntry{
			TS:     ts,
			BornAt: time.Now().UnixNano(),
			Op:     MsgSetRemove,
			Key:    key,
			Value:  []byte(mKey),
		})
	}

	if !f.disableMetrics {
		f.metrics.SetLat.Record(time.Since(start))
	}
	return removed, nil
}

// SetIsMember checks if a member is currently present in the set.
func (f *Capacitor) SetIsMember(ctx context.Context, key string, member any) (bool, error) {
	var start time.Time
	if !f.disableMetrics {
		start = time.Now()
	}
	mKey, err := memberToKey(member)
	if err != nil {
		return false, err
	}
	res, err := f.store.setIsMember(key, mKey)
	if !f.disableMetrics {
		f.metrics.GetLat.Record(time.Since(start))
	}
	return res, err
}

// SetMIsMember checks membership for multiple members in a single call.
func (f *Capacitor) SetMIsMember(ctx context.Context, key string, members ...any) ([]bool, error) {
	var start time.Time
	if !f.disableMetrics {
		start = time.Now()
	}
	res := make([]bool, len(members))
	for i, m := range members {
		mKey, err := memberToKey(m)
		if err != nil {
			return nil, err
		}
		val, err := f.store.setIsMember(key, mKey)
		if err != nil {
			return nil, err
		}
		res[i] = val
	}
	if !f.disableMetrics {
		f.metrics.GetLat.Record(time.Since(start))
	}
	return res, nil
}

// SetCard returns the cardinality (number of active members) of the set.
func (f *Capacitor) SetCard(ctx context.Context, key string) (int64, error) {
	return f.store.setCard(key)
}

// SetMembers returns all active members in the set as strings (or serialized representations).
func (f *Capacitor) SetMembers(ctx context.Context, key string) ([]string, error) {
	var start time.Time
	if !f.disableMetrics {
		start = time.Now()
	}
	raw, err := f.store.setMembers(key)
	if err != nil {
		return nil, err
	}
	res := make([]string, len(raw))
	for i, r := range raw {
		if len(r) > 0 && r[0] == '\x00' {
			res[i] = r[1:]
		} else {
			var s string
			if err := msgpack.Unmarshal([]byte(r), &s); err == nil {
				res[i] = s
			} else {
				res[i] = r
			}
		}
	}
	if !f.disableMetrics {
		f.metrics.GetLat.Record(time.Since(start))
	}
	return res, nil
}

// SetMembersScan unmarshals all active members in the set into the slice pointed to by destSlicePtr.
func (f *Capacitor) SetMembersScan(ctx context.Context, key string, destSlicePtr any) error {
	rawKeys, err := f.store.setMembers(key)
	if err != nil {
		return err
	}

	valPtr := reflect.ValueOf(destSlicePtr)
	if valPtr.Kind() != reflect.Ptr || valPtr.Elem().Kind() != reflect.Slice {
		return fmt.Errorf("destSlicePtr must be a pointer to a slice")
	}

	sliceVal := valPtr.Elem()
	elemType := sliceVal.Type().Elem()

	newSlice := reflect.MakeSlice(sliceVal.Type(), len(rawKeys), len(rawKeys))

	for i, rKey := range rawKeys {
		newElem := reflect.New(elemType)
		err := unmarshalKey(rKey, newElem.Interface())
		if err != nil {
			return err
		}
		newSlice.Index(i).Set(newElem.Elem())
	}

	sliceVal.Set(newSlice)
	return nil
}

// SetPop removes and returns a random member from the set as a string.
func (f *Capacitor) SetPop(ctx context.Context, key string) (string, error) {
	ts := f.hlc.Now()
	selectedKey, err := f.store.setPop(key, ts)
	if err != nil {
		return "", err
	}
	if selectedKey == "" {
		return "", ErrKeyNotFound
	}

	var s string
	if len(selectedKey) > 0 && selectedKey[0] == '\x00' {
		s = selectedKey[1:]
	} else {
		if err := msgpack.Unmarshal([]byte(selectedKey), &s); err != nil {
			s = selectedKey
		}
	}

	// Replicate removal
	f.log.Append(LogEntry{
		TS:     ts,
		BornAt: time.Now().UnixNano(),
		Op:     MsgSetRemove,
		Key:    key,
		Value:  []byte(selectedKey),
	})

	return s, nil
}

// SetPopScan removes a random member from the set and unmarshals it into dest.
// Returns true if a member was popped, false if the set was empty.
func (f *Capacitor) SetPopScan(ctx context.Context, key string, dest any) (bool, error) {
	valPtr := reflect.ValueOf(dest)
	if valPtr.Kind() != reflect.Ptr {
		return false, fmt.Errorf("dest must be a pointer type")
	}

	ts := f.hlc.Now()
	selectedKey, err := f.store.setPop(key, ts)
	if err != nil {
		return false, err
	}
	if selectedKey == "" {
		return false, nil
	}

	err = unmarshalKey(selectedKey, dest)
	if err != nil {
		return false, err
	}

	// Replicate removal
	f.log.Append(LogEntry{
		TS:     ts,
		BornAt: time.Now().UnixNano(),
		Op:     MsgSetRemove,
		Key:    key,
		Value:  []byte(selectedKey),
	})

	return true, nil
}

// SetRandMember returns a random member from the set as a string.
func (f *Capacitor) SetRandMember(ctx context.Context, key string) (string, error) {
	selectedKey, err := f.store.setRandMember(key)
	if err != nil {
		return "", err
	}
	if selectedKey == "" {
		return "", ErrKeyNotFound
	}
	var s string
	if len(selectedKey) > 0 && selectedKey[0] == '\x00' {
		return selectedKey[1:], nil
	}
	if err := msgpack.Unmarshal([]byte(selectedKey), &s); err != nil {
		return selectedKey, nil
	}
	return s, nil
}

// SetRandMemberScan retrieves a random member from the set and unmarshals it into dest.
// Returns true if a member was found, false if the set was empty.
func (f *Capacitor) SetRandMemberScan(ctx context.Context, key string, dest any) (bool, error) {
	valPtr := reflect.ValueOf(dest)
	if valPtr.Kind() != reflect.Ptr {
		return false, fmt.Errorf("dest must be a pointer type")
	}

	selectedKey, err := f.store.setRandMember(key)
	if err != nil {
		return false, err
	}
	if selectedKey == "" {
		return false, nil
	}

	err = unmarshalKey(selectedKey, dest)
	if err != nil {
		return false, err
	}
	return true, nil
}

// SetMove moves a member from a source set to a destination set.
func (f *Capacitor) SetMove(ctx context.Context, source string, destination string, member any) (bool, error) {
	mKey, err := memberToKey(member)
	if err != nil {
		return false, err
	}

	isMem, err := f.store.setIsMember(source, mKey)
	if err != nil {
		return false, err
	}
	if !isMem {
		return false, nil
	}

	_, err = f.SetRemove(ctx, source, member)
	if err != nil {
		return false, err
	}

	_, err = f.SetAdd(ctx, destination, member)
	if err != nil {
		return false, err
	}

	return true, nil
}

func memberToKey(member any) (string, error) {
	if s, ok := member.(string); ok {
		return "\x00" + s, nil
	}
	bin, err := msgpack.Marshal(member)
	if err != nil {
		return "", err
	}
	return string(bin), nil
}

func unmarshalKey(key string, dest any) error {
	if len(key) > 0 && key[0] == '\x00' {
		switch d := dest.(type) {
		case *string:
			*d = key[1:]
			return nil
		case *any:
			*d = key[1:]
			return nil
		}
		bin, err := msgpack.Marshal(key[1:])
		if err != nil {
			return err
		}
		return msgpack.Unmarshal(bin, dest)
	}
	return msgpack.Unmarshal([]byte(key), dest)
}

// SortedSetAdd adds a member with a score to a sorted set, or updates its score.
// Returns true if the member is newly added to the set, or false if its score was updated.
func (f *Capacitor) SortedSetAdd(ctx context.Context, key string, score float64, member any) (bool, error) {
	var start time.Time
	if !f.disableMetrics {
		start = time.Now()
	}

	mKey, err := memberToKey(member)
	if err != nil {
		return false, err
	}

	ts := f.hlc.Now()
	added, err := f.store.sortedSetAdd(key, mKey, score, ts)
	if err != nil {
		return false, err
	}

	// Replicate addition
	f.log.Append(LogEntry{
		TS:     ts,
		BornAt: time.Now().UnixNano(),
		Op:     MsgSortedSetAdd,
		Key:    key,
		Value:  []byte(mKey),
		Delta:  score,
	})

	if !f.disableMetrics {
		f.metrics.SetLat.Record(time.Since(start))
	}
	return added, nil
}

// SortedSetRemove removes a member from a sorted set.
// Returns true if the member was present and removed, or false if it was not in the set.
func (f *Capacitor) SortedSetRemove(ctx context.Context, key string, member any) (bool, error) {
	var start time.Time
	if !f.disableMetrics {
		start = time.Now()
	}

	mKey, err := memberToKey(member)
	if err != nil {
		return false, err
	}

	ts := f.hlc.Now()
	removed, err := f.store.sortedSetRemove(key, mKey, ts, false)
	if err != nil {
		return false, err
	}

	// Replicate removal
	if removed {
		f.log.Append(LogEntry{
			TS:     ts,
			BornAt: time.Now().UnixNano(),
			Op:     MsgSortedSetRemove,
			Key:    key,
			Value:  []byte(mKey),
		})
	}

	if !f.disableMetrics {
		f.metrics.SetLat.Record(time.Since(start))
	}
	return removed, nil
}

// SortedSetScore retrieves the score of a member in a sorted set.
// Returns the score and a boolean indicating whether the member is present.
func (f *Capacitor) SortedSetScore(ctx context.Context, key string, member any) (float64, bool, error) {
	mKey, err := memberToKey(member)
	if err != nil {
		return 0, false, err
	}
	return f.store.sortedSetScore(key, mKey)
}

// SortedSetCard returns the cardinality (number of active members) of a sorted set.
func (f *Capacitor) SortedSetCard(ctx context.Context, key string) (int64, error) {
	return f.store.sortedSetCard(key)
}

// SortedSetRank returns the 0-based rank of a member (ordered ascending by score).
// Returns -1 if the member is not present in the set.
func (f *Capacitor) SortedSetRank(ctx context.Context, key string, member any) (int64, error) {
	mKey, err := memberToKey(member)
	if err != nil {
		return -1, err
	}
	return f.store.sortedSetRank(key, mKey)
}

// SortedSetCount returns the number of members with a score in [min, max].
func (f *Capacitor) SortedSetCount(ctx context.Context, key string, min, max float64) (int64, error) {
	return f.store.sortedSetCount(key, min, max)
}

// SortedSetIncrementBy increments the score of a member by a delta.
// Returns the new score of the member.
func (f *Capacitor) SortedSetIncrementBy(ctx context.Context, key string, member any, delta float64) (float64, error) {
	var start time.Time
	if !f.disableMetrics {
		start = time.Now()
	}

	mKey, err := memberToKey(member)
	if err != nil {
		return 0, err
	}

	ts := f.hlc.Now()
	newScore, err := f.store.sortedSetIncrementBy(key, mKey, delta, ts)
	if err != nil {
		return 0, err
	}

	// Replicate addition
	f.log.Append(LogEntry{
		TS:     ts,
		BornAt: time.Now().UnixNano(),
		Op:     MsgSortedSetIncrement,
		Key:    key,
		Value:  []byte(mKey),
		Delta:  delta,
	})

	if !f.disableMetrics {
		f.metrics.SetLat.Record(time.Since(start))
	}
	return newScore, nil
}

// SortedSetRangeScan unmarshals a range of members (by rank index start to stop) into destSlicePtr.
// It also optionally populates destScoresPtr if not nil.
func (f *Capacitor) SortedSetRangeScan(ctx context.Context, key string, start, stop int64, destSlicePtr any, destScoresPtr *[]float64) error {
	rawKeys, scores, err := f.store.sortedSetGetRange(key, start, stop)
	if err != nil {
		return err
	}

	valPtr := reflect.ValueOf(destSlicePtr)
	if valPtr.Kind() != reflect.Ptr || valPtr.Elem().Kind() != reflect.Slice {
		return fmt.Errorf("destSlicePtr must be a pointer to a slice")
	}

	sliceVal := valPtr.Elem()
	elemType := sliceVal.Type().Elem()

	newSlice := reflect.MakeSlice(sliceVal.Type(), len(rawKeys), len(rawKeys))

	for i, rKey := range rawKeys {
		newElem := reflect.New(elemType)
		err := unmarshalKey(rKey, newElem.Interface())
		if err != nil {
			return err
		}
		newSlice.Index(i).Set(newElem.Elem())
	}

	sliceVal.Set(newSlice)

	if destScoresPtr != nil {
		*destScoresPtr = scores
	}
	return nil
}

// SortedSetRangeByScoreScan unmarshals a range of members (by score range min to max) into destSlicePtr.
// It also optionally populates destScoresPtr if not nil.
func (f *Capacitor) SortedSetRangeByScoreScan(ctx context.Context, key string, min, max float64, destSlicePtr any, destScoresPtr *[]float64) error {
	rawKeys, scores, err := f.store.sortedSetGetRangeByScore(key, min, max)
	if err != nil {
		return err
	}

	valPtr := reflect.ValueOf(destSlicePtr)
	if valPtr.Kind() != reflect.Ptr || valPtr.Elem().Kind() != reflect.Slice {
		return fmt.Errorf("destSlicePtr must be a pointer to a slice")
	}

	sliceVal := valPtr.Elem()
	elemType := sliceVal.Type().Elem()

	newSlice := reflect.MakeSlice(sliceVal.Type(), len(rawKeys), len(rawKeys))

	for i, rKey := range rawKeys {
		newElem := reflect.New(elemType)
		err := unmarshalKey(rKey, newElem.Interface())
		if err != nil {
			return err
		}
		newSlice.Index(i).Set(newElem.Elem())
	}

	sliceVal.Set(newSlice)

	if destScoresPtr != nil {
		*destScoresPtr = scores
	}
	return nil
}

// SortedSetPopScan removes a random member from the sorted set and unmarshals it into dest.
// Returns true if a member was popped and its score is returned, false if the set was empty.
func (f *Capacitor) SortedSetPopScan(ctx context.Context, key string, dest any, outScore *float64) (bool, error) {
	valPtr := reflect.ValueOf(dest)
	if valPtr.Kind() != reflect.Ptr {
		return false, fmt.Errorf("dest must be a pointer type")
	}

	ts := f.hlc.Now()
	selectedKey, score, err := f.store.sortedSetPop(key, ts)
	if err != nil {
		return false, err
	}
	if selectedKey == "" {
		return false, nil
	}

	err = unmarshalKey(selectedKey, dest)
	if err != nil {
		return false, err
	}

	// Replicate removal
	f.log.Append(LogEntry{
		TS:     ts,
		BornAt: time.Now().UnixNano(),
		Op:     MsgSortedSetRemove,
		Key:    key,
		Value:  []byte(selectedKey),
	})

	if outScore != nil {
		*outScore = score
	}
	return true, nil
}

// SortedSetRevRangeScan unmarshals a range of members in reverse score order (high-to-low) into destSlicePtr.
// It also optionally populates destScoresPtr if not nil.
func (f *Capacitor) SortedSetRevRangeScan(ctx context.Context, key string, start, stop int64, destSlicePtr any, destScoresPtr *[]float64) error {
	rawKeys, scores, err := f.store.sortedSetGetRevRange(key, start, stop)
	if err != nil {
		return err
	}

	valPtr := reflect.ValueOf(destSlicePtr)
	if valPtr.Kind() != reflect.Ptr || valPtr.Elem().Kind() != reflect.Slice {
		return fmt.Errorf("destSlicePtr must be a pointer to a slice")
	}

	sliceVal := valPtr.Elem()
	elemType := sliceVal.Type().Elem()

	newSlice := reflect.MakeSlice(sliceVal.Type(), len(rawKeys), len(rawKeys))

	for i, rKey := range rawKeys {
		newElem := reflect.New(elemType)
		err := unmarshalKey(rKey, newElem.Interface())
		if err != nil {
			return err
		}
		newSlice.Index(i).Set(newElem.Elem())
	}

	sliceVal.Set(newSlice)

	if destScoresPtr != nil {
		*destScoresPtr = scores
	}
	return nil
}

// SortedSetRevRangeByScoreScan unmarshals a range of members in reverse score order (max-to-min) into destSlicePtr.
// It also optionally populates destScoresPtr if not nil.
func (f *Capacitor) SortedSetRevRangeByScoreScan(ctx context.Context, key string, max, min float64, destSlicePtr any, destScoresPtr *[]float64) error {
	rawKeys, scores, err := f.store.sortedSetGetRevRangeByScore(key, max, min)
	if err != nil {
		return err
	}

	valPtr := reflect.ValueOf(destSlicePtr)
	if valPtr.Kind() != reflect.Ptr || valPtr.Elem().Kind() != reflect.Slice {
		return fmt.Errorf("destSlicePtr must be a pointer to a slice")
	}

	sliceVal := valPtr.Elem()
	elemType := sliceVal.Type().Elem()

	newSlice := reflect.MakeSlice(sliceVal.Type(), len(rawKeys), len(rawKeys))

	for i, rKey := range rawKeys {
		newElem := reflect.New(elemType)
		err := unmarshalKey(rKey, newElem.Interface())
		if err != nil {
			return err
		}
		newSlice.Index(i).Set(newElem.Elem())
	}

	sliceVal.Set(newSlice)

	if destScoresPtr != nil {
		*destScoresPtr = scores
	}
	return nil
}

func scanPrimitive(val string, dest any) error {
	switch d := dest.(type) {
	case *string:
		*d = val
		return nil
	case *int:
		i, err := strconv.Atoi(val)
		if err != nil {
			return err
		}
		*d = i
		return nil
	case *int64:
		i, err := strconv.ParseInt(val, 10, 64)
		if err != nil {
			return err
		}
		*d = i
		return nil
	case *float64:
		f, err := strconv.ParseFloat(val, 64)
		if err != nil {
			return err
		}
		*d = f
		return nil
	case *bool:
		b, err := strconv.ParseBool(val)
		if err != nil {
			return err
		}
		*d = b
		return nil
	case *any:
		if i, err := strconv.Atoi(val); err == nil {
			*d = i
		} else if f, err := strconv.ParseFloat(val, 64); err == nil {
			*d = f
		} else if b, err := strconv.ParseBool(val); err == nil {
			*d = b
		} else {
			*d = val
		}
		return nil
	default:
		valPtr := reflect.ValueOf(dest)
		if valPtr.Kind() != reflect.Ptr {
			return fmt.Errorf("scan target must be a pointer type")
		}
		elem := valPtr.Elem()
		switch elem.Kind() {
		case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
			i, err := strconv.ParseInt(val, 10, 64)
			if err != nil {
				return err
			}
			elem.SetInt(i)
			return nil
		case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64:
			u, err := strconv.ParseUint(val, 10, 64)
			if err != nil {
				return err
			}
			elem.SetUint(u)
			return nil
		case reflect.Float32, reflect.Float64:
			f, err := strconv.ParseFloat(val, 64)
			if err != nil {
				return err
			}
			elem.SetFloat(f)
			return nil
		case reflect.Bool:
			b, err := strconv.ParseBool(val)
			if err != nil {
				return err
			}
			elem.SetBool(b)
			return nil
		case reflect.String:
			elem.SetString(val)
			return nil
		}
		return fmt.Errorf("unsupported scan target type: %T", dest)
	}
}

// MapSet sets a field in a map to a primitive value with an optional TTL.
// Returns true if the field is newly created, false if updated.
func (f *Capacitor) MapSet(ctx context.Context, key, field string, value any, ttl time.Duration) (bool, error) {
	var start time.Time
	if !f.disableMetrics {
		start = time.Now()
	}

	valStr := valueToString(value)
	ts := f.hlc.Now()

	added, err := f.store.mapSet(key, field, valStr, ttl, ts)
	if err != nil {
		return false, err
	}

	var expiresAt int64
	if ttl > 0 {
		expiresAt = time.Now().Add(ttl).UnixNano()
	}

	payload := MapPayload{
		Field:     field,
		Value:     valStr,
		ExpiresAt: expiresAt,
	}

	bin, err := msgpack.Marshal(payload)
	if err != nil {
		return false, err
	}

	f.log.Append(LogEntry{
		TS:     ts,
		BornAt: time.Now().UnixNano(),
		Op:     MsgMapSet,
		Key:    key,
		Value:  bin,
	})

	if !f.disableMetrics {
		f.metrics.SetLat.Record(time.Since(start))
	}
	return added, nil
}

// MapGetScan retrieves a field's value and unmarshals it into destPtr.
// Returns true if found and scanned, false if the field does not exist or expired.
func (f *Capacitor) MapGetScan(ctx context.Context, key, field string, destPtr any) (bool, error) {
	valStr, found, err := f.store.mapGet(key, field)
	if err != nil || !found {
		return false, err
	}
	return true, scanPrimitive(valStr, destPtr)
}

// MapGetMScan retrieves multiple fields and unmarshals them into pointers mapped in fieldDestMap.
// Returns the number of successfully scanned fields.
func (f *Capacitor) MapGetMScan(ctx context.Context, key string, fieldDestMap map[string]any) (int, error) {
	count := 0
	for field, destPtr := range fieldDestMap {
		found, err := f.MapGetScan(ctx, key, field, destPtr)
		if err != nil {
			return count, err
		}
		if found {
			count++
		}
	}
	return count, nil
}

// MapGetAll returns a copy of the raw map containing all non-expired fields as string values.
func (f *Capacitor) MapGetAll(ctx context.Context, key string) (map[string]string, error) {
	return f.store.mapGetAll(key)
}

// MapGetAllScan retrieves all active fields of a Map and unmarshals them into destStructPtr using `map` tags.
// Returns true if the map exists and contains active fields, false otherwise.
func (f *Capacitor) MapGetAllScan(ctx context.Context, key string, destStructPtr any) (bool, error) {
	rawMap, err := f.store.mapGetAll(key)
	if err != nil || len(rawMap) == 0 {
		return false, err
	}

	valPtr := reflect.ValueOf(destStructPtr)
	if valPtr.Kind() != reflect.Ptr || valPtr.Elem().Kind() != reflect.Struct {
		return false, fmt.Errorf("destStructPtr must be a pointer to a struct")
	}

	structVal := valPtr.Elem()
	structType := structVal.Type()

	for i := 0; i < structVal.NumField(); i++ {
		field := structType.Field(i)
		tag := field.Tag.Get("map")
		if tag == "" || tag == "-" {
			continue
		}

		if valStr, ok := rawMap[tag]; ok {
			fieldVal := structVal.Field(i)
			if fieldVal.CanSet() {
				err := scanPrimitive(valStr, fieldVal.Addr().Interface())
				if err != nil {
					return false, err
				}
			}
		}
	}

	return true, nil
}

// MapIncrementBy increments a numeric field by a delta.
// Returns the final score.
func (f *Capacitor) MapIncrementBy(ctx context.Context, key, field string, delta float64) (float64, error) {
	var start time.Time
	if !f.disableMetrics {
		start = time.Now()
	}

	ts := f.hlc.Now()
	newVal, err := f.store.mapIncrementBy(key, field, delta, ts)
	if err != nil {
		return 0, err
	}

	valStr := strconv.FormatFloat(newVal, 'f', -1, 64)
	payload := MapPayload{
		Field: field,
		Value: valStr,
	}

	bin, err := msgpack.Marshal(payload)
	if err != nil {
		return 0, err
	}

	f.log.Append(LogEntry{
		TS:     ts,
		BornAt: time.Now().UnixNano(),
		Op:     MsgMapSet,
		Key:    key,
		Value:  bin,
	})

	if !f.disableMetrics {
		f.metrics.SetLat.Record(time.Since(start))
	}
	return newVal, nil
}

// MapSetTTL sets a field-level TTL on a map.
func (f *Capacitor) MapSetTTL(ctx context.Context, key, field string, ttl time.Duration) (bool, error) {
	ts := f.hlc.Now()
	ok, err := f.store.mapSetTTL(key, field, ttl, ts)
	if err != nil || !ok {
		return false, err
	}

	var expiresAt int64
	if ttl > 0 {
		expiresAt = time.Now().Add(ttl).UnixNano()
	}

	payload := MapPayload{
		Field:     field,
		ExpiresAt: expiresAt,
	}
	bin, _ := msgpack.Marshal(payload)

	f.log.Append(LogEntry{
		TS:     ts,
		BornAt: time.Now().UnixNano(),
		Op:     MsgMapSet,
		Key:    key,
		Value:  bin,
	})
	return true, nil
}

// MapSetExpiry sets a field-level expiry timestamp in Unix milliseconds.
func (f *Capacitor) MapSetExpiry(ctx context.Context, key, field string, expireTimeMs int64) (bool, error) {
	ttl := time.Duration(expireTimeMs-time.Now().UnixMilli()) * time.Millisecond
	if ttl < 0 {
		ttl = 1 * time.Nanosecond
	}
	return f.MapSetTTL(ctx, key, field, ttl)
}

// MapGetTTL retrieves the remaining TTL duration of a field.
func (f *Capacitor) MapGetTTL(ctx context.Context, key, field string) (time.Duration, bool, error) {
	return f.store.mapGetTTL(key, field)
}

// MapGetExpiry retrieves the expiry Unix timestamp in milliseconds of a field.
func (f *Capacitor) MapGetExpiry(ctx context.Context, key, field string) (int64, bool, error) {
	ttl, ok, err := f.store.mapGetTTL(key, field)
	if err != nil || !ok {
		return 0, false, err
	}
	if ttl == NoExpiry {
		return 0, true, nil
	}
	return time.Now().Add(ttl).UnixMilli(), true, nil
}

// MapRemoveTTL removes the TTL of a field, converting it to persistent.
func (f *Capacitor) MapRemoveTTL(ctx context.Context, key, field string) (bool, error) {
	ts := f.hlc.Now()
	ok, err := f.store.mapRemoveTTL(key, field, ts)
	if err != nil || !ok {
		return false, err
	}

	payload := MapPayload{
		Field:     field,
		ExpiresAt: 0,
	}
	bin, _ := msgpack.Marshal(payload)

	f.log.Append(LogEntry{
		TS:     ts,
		BornAt: time.Now().UnixNano(),
		Op:     MsgMapSet,
		Key:    key,
		Value:  bin,
	})
	return true, nil
}

// MapRemoveExpiry is an alias to MapRemoveTTL.
func (f *Capacitor) MapRemoveExpiry(ctx context.Context, key, field string) (bool, error) {
	return f.MapRemoveTTL(ctx, key, field)
}

// MapRemove deletes a field from a map.
func (f *Capacitor) MapRemove(ctx context.Context, key, field string) (bool, error) {
	var start time.Time
	if !f.disableMetrics {
		start = time.Now()
	}

	ts := f.hlc.Now()
	removed, err := f.store.mapRemove(key, field, ts)
	if err != nil || !removed {
		return false, err
	}

	payload := MapPayload{
		Field: field,
	}
	bin, _ := msgpack.Marshal(payload)

	f.log.Append(LogEntry{
		TS:     ts,
		BornAt: time.Now().UnixNano(),
		Op:     MsgMapRemove,
		Key:    key,
		Value:  bin,
	})

	if !f.disableMetrics {
		f.metrics.SetLat.Record(time.Since(start))
	}
	return true, nil
}

// MapExists returns true if a field exists and is active in a Map.
func (f *Capacitor) MapExists(ctx context.Context, key, field string) (bool, error) {
	return f.store.mapExists(key, field)
}

// MapLen returns the count of active fields in a Map.
func (f *Capacitor) MapLen(ctx context.Context, key string) (int, error) {
	return f.store.mapLen(key)
}

// MapKeysScan unmarshals all active field keys of a Map into destSlicePtr.
func (f *Capacitor) MapKeysScan(ctx context.Context, key string, destSlicePtr any) error {
	keys, err := f.store.mapKeys(key)
	if err != nil {
		return err
	}

	valPtr := reflect.ValueOf(destSlicePtr)
	if valPtr.Kind() != reflect.Ptr || valPtr.Elem().Kind() != reflect.Slice {
		return fmt.Errorf("destSlicePtr must be a pointer to a slice")
	}

	sliceVal := valPtr.Elem()
	elemType := sliceVal.Type().Elem()

	newSlice := reflect.MakeSlice(sliceVal.Type(), len(keys), len(keys))
	for i, k := range keys {
		newElem := reflect.New(elemType)
		err := scanPrimitive(k, newElem.Interface())
		if err != nil {
			return err
		}
		newSlice.Index(i).Set(newElem.Elem())
	}

	sliceVal.Set(newSlice)
	return nil
}

// MapValuesScan unmarshals all active field values of a Map into destSlicePtr.
func (f *Capacitor) MapValuesScan(ctx context.Context, key string, destSlicePtr any) error {
	vals, err := f.store.mapValues(key)
	if err != nil {
		return err
	}

	valPtr := reflect.ValueOf(destSlicePtr)
	if valPtr.Kind() != reflect.Ptr || valPtr.Elem().Kind() != reflect.Slice {
		return fmt.Errorf("destSlicePtr must be a pointer to a slice")
	}

	sliceVal := valPtr.Elem()
	elemType := sliceVal.Type().Elem()

	newSlice := reflect.MakeSlice(sliceVal.Type(), len(vals), len(vals))
	for i, v := range vals {
		newElem := reflect.New(elemType)
		err := scanPrimitive(v, newElem.Interface())
		if err != nil {
			return err
		}
		newSlice.Index(i).Set(newElem.Elem())
	}

	sliceVal.Set(newSlice)
	return nil
}

// NMapSet sets a field in a nested map to any value (nested map/struct/slice/primitive) with an optional TTL.
// Returns true if the field is newly created, false if updated.
func (f *Capacitor) NMapSet(ctx context.Context, key, field string, value any, ttl time.Duration) (bool, error) {
	var start time.Time
	if !f.disableMetrics {
		start = time.Now()
	}

	valBin, err := msgpack.Marshal(value)
	if err != nil {
		return false, err
	}

	ts := f.hlc.Now()
	rootKey, rootVal, added, err := f.store.nmapSet(key, field, valBin, ttl, ts, false)
	if err != nil {
		return false, err
	}

	var expiresAt int64
	if ttl > 0 {
		expiresAt = time.Now().Add(ttl).UnixNano()
	}

	payload := NMapPayload{
		Field:     rootKey,
		Value:     rootVal,
		ExpiresAt: expiresAt,
	}

	bin, err := msgpack.Marshal(payload)
	if err != nil {
		return false, err
	}

	f.log.Append(LogEntry{
		TS:     ts,
		BornAt: time.Now().UnixNano(),
		Op:     MsgNMapSet,
		Key:    key,
		Value:  bin,
	})

	if !f.disableMetrics {
		f.metrics.SetLat.Record(time.Since(start))
	}
	return added, nil
}

// NMapGetScan retrieves a field's value (primitive or nested structure) and unmarshals it into destPtr.
// Returns true if found and scanned, false if the field does not exist or expired.
func (f *Capacitor) NMapGetScan(ctx context.Context, key, field string, destPtr any) (bool, error) {
	valBin, found, err := f.store.nmapGet(key, field)
	if err != nil || !found {
		return false, err
	}
	err = msgpack.Unmarshal(valBin, destPtr)
	return err == nil, err
}

// NMapGetMScan retrieves multiple fields and unmarshals them into pointers mapped in fieldDestMap.
// Returns the number of successfully scanned fields.
func (f *Capacitor) NMapGetMScan(ctx context.Context, key string, fieldDestMap map[string]any) (int, error) {
	count := 0
	for field, destPtr := range fieldDestMap {
		found, err := f.NMapGetScan(ctx, key, field, destPtr)
		if err != nil {
			return count, err
		}
		if found {
			count++
		}
	}
	return count, nil
}

// NMapGetAll retrieves all active fields of an NMap as deserialized values.
func (f *Capacitor) NMapGetAll(ctx context.Context, key string) (map[string]any, error) {
	rawMap, err := f.store.nmapGetAll(key)
	if err != nil {
		return nil, err
	}

	res := make(map[string]any)
	for k, vBin := range rawMap {
		var val any
		if err := msgpack.Unmarshal(vBin, &val); err == nil {
			res[k] = val
		}
	}
	return res, nil
}

// NMapGetAllScan retrieves all active fields of an NMap and unmarshals them into destStructPtr using `map` tags.
// Returns true if the map exists and contains active fields, false otherwise.
func (f *Capacitor) NMapGetAllScan(ctx context.Context, key string, destStructPtr any) (bool, error) {
	rawMap, err := f.store.nmapGetAll(key)
	if err != nil || len(rawMap) == 0 {
		return false, err
	}

	valPtr := reflect.ValueOf(destStructPtr)
	if valPtr.Kind() != reflect.Ptr || valPtr.Elem().Kind() != reflect.Struct {
		return false, fmt.Errorf("destStructPtr must be a pointer to a struct")
	}

	structVal := valPtr.Elem()
	structType := structVal.Type()

	for i := 0; i < structVal.NumField(); i++ {
		field := structType.Field(i)
		tag := field.Tag.Get("map")
		if tag == "" || tag == "-" {
			continue
		}

		if valBin, ok := rawMap[tag]; ok {
			fieldVal := structVal.Field(i)
			if fieldVal.CanSet() {
				_ = msgpack.Unmarshal(valBin, fieldVal.Addr().Interface())
			}
		}
	}

	return true, nil
}

// NMapIncrementBy increments a numeric field by a delta.
// Returns the final score.
func (f *Capacitor) NMapIncrementBy(ctx context.Context, key, field string, delta float64) (float64, error) {
	var start time.Time
	if !f.disableMetrics {
		start = time.Now()
	}

	ts := f.hlc.Now()
	rootKey, rootVal, newVal, err := f.store.nmapIncrementBy(key, field, delta, ts)
	if err != nil {
		return 0, err
	}

	payload := NMapPayload{
		Field: rootKey,
		Value: rootVal,
	}

	bin, err := msgpack.Marshal(payload)
	if err != nil {
		return 0, err
	}

	f.log.Append(LogEntry{
		TS:     ts,
		BornAt: time.Now().UnixNano(),
		Op:     MsgNMapSet,
		Key:    key,
		Value:  bin,
	})

	if !f.disableMetrics {
		f.metrics.SetLat.Record(time.Since(start))
	}
	return newVal, nil
}

// NMapSetTTL sets a field-level TTL on an NMap.
func (f *Capacitor) NMapSetTTL(ctx context.Context, key, field string, ttl time.Duration) (bool, error) {
	ts := f.hlc.Now()
	rootKey, ok, err := f.store.nmapSetTTL(key, field, ttl, ts)
	if err != nil || !ok {
		return false, err
	}

	var expiresAt int64
	if ttl > 0 {
		expiresAt = time.Now().Add(ttl).UnixNano()
	}

	payload := NMapPayload{
		Field:     rootKey,
		ExpiresAt: expiresAt,
	}
	bin, _ := msgpack.Marshal(payload)

	f.log.Append(LogEntry{
		TS:     ts,
		BornAt: time.Now().UnixNano(),
		Op:     MsgNMapSet,
		Key:    key,
		Value:  bin,
	})
	return true, nil
}

// NMapSetExpiry sets a field-level expiry timestamp in Unix milliseconds.
func (f *Capacitor) NMapSetExpiry(ctx context.Context, key, field string, expireTimeMs int64) (bool, error) {
	ttl := time.Duration(expireTimeMs-time.Now().UnixMilli()) * time.Millisecond
	if ttl < 0 {
		ttl = 1 * time.Nanosecond
	}
	return f.NMapSetTTL(ctx, key, field, ttl)
}

// NMapGetTTL retrieves the remaining TTL duration of a field.
func (f *Capacitor) NMapGetTTL(ctx context.Context, key, field string) (time.Duration, bool, error) {
	return f.store.nmapGetTTL(key, field)
}

// NMapGetExpiry retrieves the expiry Unix timestamp in milliseconds of a field.
func (f *Capacitor) NMapGetExpiry(ctx context.Context, key, field string) (int64, bool, error) {
	ttl, ok, err := f.store.nmapGetTTL(key, field)
	if err != nil || !ok {
		return 0, false, err
	}
	if ttl == NoExpiry {
		return 0, true, nil
	}
	return time.Now().Add(ttl).UnixMilli(), true, nil
}

// NMapRemoveTTL removes the TTL of a field, converting it to persistent.
func (f *Capacitor) NMapRemoveTTL(ctx context.Context, key, field string) (bool, error) {
	ts := f.hlc.Now()
	rootKey, ok, err := f.store.nmapRemoveTTL(key, field, ts)
	if err != nil || !ok {
		return false, err
	}

	payload := NMapPayload{
		Field:     rootKey,
		ExpiresAt: 0,
	}
	bin, _ := msgpack.Marshal(payload)

	f.log.Append(LogEntry{
		TS:     ts,
		BornAt: time.Now().UnixNano(),
		Op:     MsgNMapSet,
		Key:    key,
		Value:  bin,
	})
	return true, nil
}

// NMapRemoveExpiry is an alias to NMapRemoveTTL.
func (f *Capacitor) NMapRemoveExpiry(ctx context.Context, key, field string) (bool, error) {
	return f.NMapRemoveTTL(ctx, key, field)
}

// NMapRemove deletes a field from an NMap.
func (f *Capacitor) NMapRemove(ctx context.Context, key, field string) (bool, error) {
	var start time.Time
	if !f.disableMetrics {
		start = time.Now()
	}

	ts := f.hlc.Now()
	rootKey, rootVal, isRootRemoved, removed, err := f.store.nmapRemove(key, field, ts, false)
	if err != nil || !removed {
		return false, err
	}

	var op MsgType
	var payload NMapPayload
	if isRootRemoved {
		op = MsgNMapRemove
		payload = NMapPayload{
			Field: rootKey,
		}
	} else {
		op = MsgNMapSet
		payload = NMapPayload{
			Field: rootKey,
			Value: rootVal,
		}
	}

	bin, _ := msgpack.Marshal(payload)

	f.log.Append(LogEntry{
		TS:     ts,
		BornAt: time.Now().UnixNano(),
		Op:     op,
		Key:    key,
		Value:  bin,
	})

	if !f.disableMetrics {
		f.metrics.SetLat.Record(time.Since(start))
	}
	return true, nil
}

// NMapExists returns true if a field exists and is active in an NMap.
func (f *Capacitor) NMapExists(ctx context.Context, key, field string) (bool, error) {
	return f.store.nmapExists(key, field)
}

// NMapLen returns the count of active fields in an NMap.
func (f *Capacitor) NMapLen(ctx context.Context, key string) (int, error) {
	return f.store.nmapLen(key)
}

// NMapKeysScan unmarshals all active field keys of an NMap into destSlicePtr.
func (f *Capacitor) NMapKeysScan(ctx context.Context, key string, destSlicePtr any) error {
	keys, err := f.store.nmapKeys(key)
	if err != nil {
		return err
	}

	valPtr := reflect.ValueOf(destSlicePtr)
	if valPtr.Kind() != reflect.Ptr || valPtr.Elem().Kind() != reflect.Slice {
		return fmt.Errorf("destSlicePtr must be a pointer to a slice")
	}

	sliceVal := valPtr.Elem()
	elemType := sliceVal.Type().Elem()

	newSlice := reflect.MakeSlice(sliceVal.Type(), len(keys), len(keys))
	for i, k := range keys {
		newElem := reflect.New(elemType)
		err := scanPrimitive(k, newElem.Interface())
		if err != nil {
			return err
		}
		newSlice.Index(i).Set(newElem.Elem())
	}

	sliceVal.Set(newSlice)
	return nil
}

// NMapValuesScan unmarshals all active field values of an NMap into destSlicePtr.
func (f *Capacitor) NMapValuesScan(ctx context.Context, key string, destSlicePtr any) error {
	vals, err := f.store.nmapValues(key)
	if err != nil {
		return err
	}

	valPtr := reflect.ValueOf(destSlicePtr)
	if valPtr.Kind() != reflect.Ptr || valPtr.Elem().Kind() != reflect.Slice {
		return fmt.Errorf("destSlicePtr must be a pointer to a slice")
	}

	sliceVal := valPtr.Elem()
	elemType := sliceVal.Type().Elem()

	newSlice := reflect.MakeSlice(sliceVal.Type(), len(vals), len(vals))
	for i, vBin := range vals {
		newElem := reflect.New(elemType)
		err := msgpack.Unmarshal(vBin, newElem.Interface())
		if err != nil {
			return err
		}
		newSlice.Index(i).Set(newElem.Elem())
	}

	sliceVal.Set(newSlice)
	return nil
}

// ListLeftPush prepends a value to the list.
func (f *Capacitor) ListLeftPush(ctx context.Context, key string, value any) error {
	var start time.Time
	if !f.disableMetrics {
		start = time.Now()
	}

	valBin, err := msgpack.Marshal(value)
	if err != nil {
		return err
	}

	ts := f.hlc.Now()
	_, err = f.store.listInsert(key, ts, Timestamp{}, valBin, ts)
	if err != nil {
		return err
	}

	payload := ListPayload{
		ID:       ts,
		ParentID: Timestamp{},
		Value:    valBin,
	}
	bin, err := msgpack.Marshal(payload)
	if err != nil {
		return err
	}

	f.log.Append(LogEntry{
		TS:     ts,
		BornAt: time.Now().UnixNano(),
		Op:     MsgListInsert,
		Key:    key,
		Value:  bin,
	})

	if !f.disableMetrics {
		f.metrics.SetLat.Record(time.Since(start))
	}
	return nil
}

// ListRightPush appends a value to the list.
func (f *Capacitor) ListRightPush(ctx context.Context, key string, value any) error {
	var start time.Time
	if !f.disableMetrics {
		start = time.Now()
	}

	valBin, err := msgpack.Marshal(value)
	if err != nil {
		return err
	}

	flat, err := f.store.listFlatten(key)
	if err != nil {
		return err
	}

	var parentID Timestamp
	if len(flat) > 0 {
		parentID = flat[len(flat)-1].ID
	}

	ts := f.hlc.Now()
	_, err = f.store.listInsert(key, ts, parentID, valBin, ts)
	if err != nil {
		return err
	}

	payload := ListPayload{
		ID:       ts,
		ParentID: parentID,
		Value:    valBin,
	}
	bin, err := msgpack.Marshal(payload)
	if err != nil {
		return err
	}

	f.log.Append(LogEntry{
		TS:     ts,
		BornAt: time.Now().UnixNano(),
		Op:     MsgListInsert,
		Key:    key,
		Value:  bin,
	})

	if !f.disableMetrics {
		f.metrics.SetLat.Record(time.Since(start))
	}
	return nil
}

// ListLeftPopScan pops and returns the first element from the head of the list, unmarshaling it into destPtr.
func (f *Capacitor) ListLeftPopScan(ctx context.Context, key string, destPtr any) (bool, error) {
	flat, err := f.store.listFlatten(key)
	if err != nil || len(flat) == 0 {
		return false, err
	}

	target := flat[0]
	err = msgpack.Unmarshal(target.Value, destPtr)
	if err != nil {
		return false, err
	}

	ts := f.hlc.Now()
	_, err = f.store.listDelete(key, target.ID, ts)
	if err != nil {
		return false, err
	}

	payload := ListPayload{
		ID:        target.ID,
		DeletedAt: ts,
	}
	bin, _ := msgpack.Marshal(payload)

	f.log.Append(LogEntry{
		TS:     ts,
		BornAt: time.Now().UnixNano(),
		Op:     MsgListDelete,
		Key:    key,
		Value:  bin,
	})

	return true, nil
}

// ListRightPopScan pops and returns the last element from the tail of the list, unmarshaling it into destPtr.
func (f *Capacitor) ListRightPopScan(ctx context.Context, key string, destPtr any) (bool, error) {
	flat, err := f.store.listFlatten(key)
	if err != nil || len(flat) == 0 {
		return false, err
	}

	target := flat[len(flat)-1]
	err = msgpack.Unmarshal(target.Value, destPtr)
	if err != nil {
		return false, err
	}

	ts := f.hlc.Now()
	_, err = f.store.listDelete(key, target.ID, ts)
	if err != nil {
		return false, err
	}

	payload := ListPayload{
		ID:        target.ID,
		DeletedAt: ts,
	}
	bin, _ := msgpack.Marshal(payload)

	f.log.Append(LogEntry{
		TS:     ts,
		BornAt: time.Now().UnixNano(),
		Op:     MsgListDelete,
		Key:    key,
		Value:  bin,
	})

	return true, nil
}

// ListLen returns the length of the list.
func (f *Capacitor) ListLen(ctx context.Context, key string) (int, error) {
	flat, err := f.store.listFlatten(key)
	if err != nil {
		return 0, err
	}
	return len(flat), nil
}

// ListRangeScan retrieves a sub-range of elements and unmarshals them into destSlicePtr.
func (f *Capacitor) ListRangeScan(ctx context.Context, key string, start, stop int, destSlicePtr any) error {
	flat, err := f.store.listFlatten(key)
	if err != nil {
		return err
	}

	n := len(flat)
	if n == 0 {
		return nil
	}

	if start < 0 {
		start = n + start
	}
	if stop < 0 {
		stop = n + stop
	}

	if start < 0 {
		start = 0
	}
	if stop >= n {
		stop = n - 1
	}

	if start > stop || start >= n {
		return nil
	}

	sliceRange := flat[start : stop+1]

	valPtr := reflect.ValueOf(destSlicePtr)
	if valPtr.Kind() != reflect.Ptr || valPtr.Elem().Kind() != reflect.Slice {
		return fmt.Errorf("destSlicePtr must be a pointer to a slice")
	}

	sliceVal := valPtr.Elem()
	elemType := sliceVal.Type().Elem()

	newSlice := reflect.MakeSlice(sliceVal.Type(), len(sliceRange), len(sliceRange))
	for i, item := range sliceRange {
		newElem := reflect.New(elemType)
		err := msgpack.Unmarshal(item.Value, newElem.Interface())
		if err != nil {
			return err
		}
		newSlice.Index(i).Set(newElem.Elem())
	}

	sliceVal.Set(newSlice)
	return nil
}

// HLLAdd adds one or more elements to the HyperLogLog cardinality estimator.
func (f *Capacitor) HLLAdd(ctx context.Context, key string, elements ...string) (bool, error) {
	ts := f.hlc.Now()
	anyUpdated := false

	for _, elem := range elements {
		hash := murmurHash64([]byte(elem))
		idx := uint32(hash & 1023)
		val := byte(bits.LeadingZeros64(hash>>10) + 1)

		updated, err := f.store.hllAdd(key, idx, val, ts)
		if err != nil {
			return false, err
		}
		if updated {
			anyUpdated = true
			payload := HLLPayload{
				Index: idx,
				Value: val,
			}
			bin, _ := msgpack.Marshal(payload)
			f.log.Append(LogEntry{
				TS:     ts,
				BornAt: time.Now().UnixNano(),
				Op:     MsgHLLAdd,
				Key:    key,
				Value:  bin,
			})
		}
	}

	return anyUpdated, nil
}

// HLLCount returns the estimated cardinality of the HyperLogLog structure.
func (f *Capacitor) HLLCount(ctx context.Context, key string) (int64, error) {
	return f.store.hllCount(key)
}

// BloomAdd inserts an element into the Bloom Filter.
func (f *Capacitor) BloomAdd(ctx context.Context, key string, element string) (bool, error) {
	ts := f.hlc.Now()

	hash := murmurHash64([]byte(element))
	h1 := uint32(hash)
	h2 := uint32(hash >> 32)

	indices := make([]uint32, 4)
	for i := uint32(0); i < 4; i++ {
		indices[i] = (h1 + i*h2) % 8192
	}

	updated, err := f.store.bloomAdd(key, indices, ts)
	if err != nil {
		return false, err
	}

	if updated {
		payload := BloomPayload{
			Indices: indices,
		}
		bin, _ := msgpack.Marshal(payload)
		f.log.Append(LogEntry{
			TS:     ts,
			BornAt: time.Now().UnixNano(),
			Op:     MsgBloomAdd,
			Key:    key,
			Value:  bin,
		})
	}

	return updated, nil
}

// BloomExists checks if the element is likely present in the Bloom Filter.
func (f *Capacitor) BloomExists(ctx context.Context, key string, element string) (bool, error) {
	hash := murmurHash64([]byte(element))
	h1 := uint32(hash)
	h2 := uint32(hash >> 32)

	indices := make([]uint32, 4)
	for i := uint32(0); i < 4; i++ {
		indices[i] = (h1 + i*h2) % 8192
	}

	return f.store.bloomExists(key, indices)
}

// CMSIncrement increments the count of an element in the Count-Min Sketch grid.
func (f *Capacitor) CMSIncrement(ctx context.Context, key string, element string, count int64) error {
	if count < 0 {
		return errors.New("count cannot be negative")
	}

	ts := f.hlc.Now()

	err := f.store.cmsIncrement(key, element, count, ts)
	if err != nil {
		return err
	}

	payload := CMSPayload{
		Element: element,
		Count:   count,
	}
	bin, _ := msgpack.Marshal(payload)

	f.log.Append(LogEntry{
		TS:     ts,
		BornAt: time.Now().UnixNano(),
		Op:     MsgCMSIncrement,
		Key:    key,
		Value:  bin,
	})

	return nil
}

// CMSQuery returns the estimated frequency of the element in the Count-Min Sketch grid.
func (f *Capacitor) CMSQuery(ctx context.Context, key string, element string) (int64, error) {
	return f.store.cmsQuery(key, element)
}

type subscriptionController struct {
	closeCh chan struct{}
	once    sync.Once
	cleanup func()
}

func (c *subscriptionController) unsubscribe() {
	c.once.Do(func() {
		close(c.closeCh)
		c.cleanup()
	})
}

type Subscription struct {
	Channel <-chan PubSubMessage
	ctrl    *subscriptionController
}

func (s *Subscription) Unsubscribe() {
	s.ctrl.unsubscribe()
}

func (m *PubSubMessage) Scan(destPtr any) error {
	return msgpack.Unmarshal(m.Payload, destPtr)
}

// Subscribe registers a subscriber Go channel for a topic.
func (f *Capacitor) Subscribe(ctx context.Context, topic string) (*Subscription, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}

	id := atomic.AddUint64(&f.subCounter, 1)
	subID := strconv.FormatUint(id, 10)

	ch := make(chan PubSubMessage, 256)
	f.pubSub.subscribe(topic, subID, ch)
	atomic.AddInt64(&f.metrics.ActiveSubscriptions, 1)

	closeCh := make(chan struct{})
	ctrl := &subscriptionController{
		closeCh: closeCh,
		cleanup: func() {
			f.pubSub.unsubscribe(topic, subID)
			atomic.AddInt64(&f.metrics.ActiveSubscriptions, -1)
			close(ch)
		},
	}

	sub := &Subscription{
		Channel: ch,
		ctrl:    ctrl,
	}

	go func() {
		select {
		case <-ctx.Done():
			ctrl.unsubscribe()
		case <-f.ctx.Done():
			ctrl.unsubscribe()
		case <-closeCh:
		}
	}()

	runtime.SetFinalizer(sub, func(s *Subscription) {
		s.Unsubscribe()
	})

	return sub, nil
}

// Publish distributes a message payload to all subscribers of a topic across the cluster.
func (f *Capacitor) Publish(ctx context.Context, topic string, message any) error {
	valBin, err := msgpack.Marshal(message)
	if err != nil {
		return err
	}

	f.pubSub.publish(topic, PubSubMessage{Topic: topic, Payload: valBin})

	payload := PubSubPayload{
		Topic:   topic,
		Payload: valBin,
	}
	bin, _ := msgpack.Marshal(payload)

	f.log.Append(LogEntry{
		TS:     f.hlc.Now(),
		BornAt: time.Now().UnixNano(),
		Op:     MsgPubSubPublish,
		Key:    topic,
		Value:  bin,
	})

	return nil
}

// Close gracefully shuts down the node, leaving the SWIM gossip group, shutting down
// active replication streams, and closing local storage engines.
func (f *Capacitor) Close() error {
	atomic.StoreUint32(&f.closing, 1)

	if f.ml != nil {
		f.ml.Leave(2 * time.Second)
		f.ml.Shutdown()
	}

	f.server.Stop()

	f.log.mu.RLock()
	targetSeq := f.log.head - 1
	f.log.mu.RUnlock()

	flushCtx, flushCancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer flushCancel()

	ticker := time.NewTicker(20 * time.Millisecond)
	defer ticker.Stop()

Outer:
	for {
		select {
		case <-flushCtx.Done():
			break Outer
		case <-ticker.C:
			allCaughtUp := true
			f.peerMutex.Lock()
			for nodeID := range f.peerStops {
				lastSeqVal, ok := f.peerSeqs.Load(nodeID)
				var lastSeq uint64
				if ok && lastSeqVal != nil {
					lastSeq = lastSeqVal.(uint64)
				}
				if lastSeq < targetSeq {
					allCaughtUp = false
					break
				}
			}
			f.peerMutex.Unlock()
			if allCaughtUp {
				break Outer
			}
		}
	}

	f.cancel()
	close(f.stop)

	f.peerMutex.Lock()
	var nodesToStop []string
	for nodeID := range f.peerStops {
		nodesToStop = append(nodesToStop, nodeID)
	}
	f.peerMutex.Unlock()

	for _, nodeID := range nodesToStop {
		f.stopPeerReplicator(nodeID)
	}

	f.log.mu.Lock()
	close(f.log.notifyCh)
	f.log.notifyCh = make(chan struct{})
	f.log.mu.Unlock()

	f.replicatorWG.Wait()

	f.metrics.Stop()
	f.client.Close()
	return f.store.close()
}

func valueToString(value any) string {
	switch v := value.(type) {
	case string:
		return v
	case []byte:
		return string(v)
	case int:
		return strconv.Itoa(v)
	case int64:
		return strconv.FormatInt(v, 10)
	case int32:
		return strconv.FormatInt(int64(v), 10)
	case int16:
		return strconv.FormatInt(int64(v), 10)
	case int8:
		return strconv.FormatInt(int64(v), 10)
	case uint:
		return strconv.FormatUint(uint64(v), 10)
	case uint64:
		return strconv.FormatUint(v, 10)
	case uint32:
		return strconv.FormatUint(uint64(v), 10)
	case uint16:
		return strconv.FormatUint(uint64(v), 10)
	case uint8:
		return strconv.FormatUint(uint64(v), 10)
	case float64:
		return strconv.FormatFloat(v, 'g', -1, 64)
	case float32:
		return strconv.FormatFloat(float64(v), 'g', -1, 32)
	case bool:
		return strconv.FormatBool(v)
	default:
		return fmt.Sprintf("%v", value)
	}
}

func getLocalIP() string {
	addrs, err := net.InterfaceAddrs()
	if err != nil {
		return "127.0.0.1"
	}
	for _, addr := range addrs {
		if ipnet, ok := addr.(*net.IPNet); ok && !ipnet.IP.IsLoopback() {
			if ipnet.IP.To4() != nil {
				return ipnet.IP.String()
			}
		}
	}
	return "127.0.0.1"
}

func isRunningTests() bool {
	for _, arg := range os.Args {
		if strings.HasPrefix(arg, "-test.") {
			return true
		}
	}
	return strings.HasSuffix(os.Args[0], ".test")
}