libkvstore

README

libkvstore

A small, backend-agnostic key-value store abstraction for Go. It defines a single KVExecutor interface — get/set/delete/exists/keys plus list and set operations, all values as json.RawMessage — and ships two implementations:

• SQLite, built on top of libdbexec's DBManager. Lists and sets are stored as a JSON array in a single row; read-modify-write operations (ListPush, ListTrim, ListRPop, SetAdd, SetRemove) run inside a write-locked transaction (via a no-op write that promotes SQLite's DEFERRED transaction immediately) with retry-on-busy, so concurrent callers don't lose updates.
• Valkey/Redis, using valkey-io/valkey-go. Operations map directly onto the equivalent Valkey commands (GET/SET, LPUSH/LRANGE/LTRIM/LLEN/RPOP, SADD/SMEMBERS/SREM), with an optional default TTL applied to writes.

Both implementations share the same ErrNotFound, ErrInvalidValue, ErrKeyExists, ErrTxFailed, ErrTxConflict, ErrLockUnavailable, and ErrConnectionFailed sentinel errors, so callers can be written against the interface without caring which backend is behind it.

local.go and local_valkey.go provide testcontainers-based helpers (SetupLocalValKeyInstance, SetupLocalPDInstance, SetupLocalTiKVInstance) for spinning up a real Valkey or PD/TiKV instance in tests.

Install

go get github.com/contenox/libkvstore

Usage

SQLite backend

db, err := libdbexec.NewSQLiteDBManager(ctx, "path/to/kv.db", libkvstore.SQLiteSchema)
if err != nil {
    // handle error
}
defer db.Close()

mgr := libkvstore.NewSQLiteManager(db)
kv, err := mgr.Executor(ctx)
if err != nil {
    // handle error
}

if err := kv.Set(ctx, "key", json.RawMessage(`"value"`)); err != nil {
    // handle error
}

Valkey backend

mgr, err := libkvstore.NewManager(libkvstore.Config{
    KVAddr: "localhost:6379",
}, 0 /* no default TTL */)
if err != nil {
    // handle error
}
defer mgr.Close()

kv, err := mgr.Executor(ctx)
if err != nil {
    // handle error
}

Dependency on libdbexec

The SQLite backend (sqlite.go) is built directly on github.com/contenox/libdbexec's DBManager/Exec types — it opens transactions and issues SQL through that package rather than talking to database/sql itself. libdbexec is a sibling module; see go.mod for how it's wired in.

License

Apache-2.0, see LICENSE.

Documentation

Index

• Constants
• Variables
• func SetupLocalPDInstance(ctx context.Context) (string, testcontainers.Container, func(), error)
• func SetupLocalTiKVInstance(ctx context.Context) (string, testcontainers.Container, func(), error)
• func SetupLocalValKeyInstance(ctx context.Context) (string, testcontainers.Container, func(), error)
• type Config
• type KVExecutor
• type KVManager
• type Key
• type KeyValue
• type SQLiteManager
  • func NewSQLiteManager(db libdbexec.DBManager) *SQLiteManager
  • func (m *SQLiteManager) Close() error
  • func (m *SQLiteManager) Executor(_ context.Context) (KVExecutor, error)
• type VKExecutor
  • func (r *VKExecutor) Delete(ctx context.Context, key Key) error
  • func (r *VKExecutor) Exists(ctx context.Context, key Key) (bool, error)
  • func (r *VKExecutor) Get(ctx context.Context, key Key) (json.RawMessage, error)
  • func (r *VKExecutor) Keys(ctx context.Context, pattern string) ([]Key, error)
  • func (r *VKExecutor) ListLength(ctx context.Context, key Key) (int64, error)
  • func (r *VKExecutor) ListPush(ctx context.Context, key Key, value json.RawMessage) error
  • func (r *VKExecutor) ListRPop(ctx context.Context, key Key) (json.RawMessage, error)
  • func (r *VKExecutor) ListRange(ctx context.Context, key Key, start, stop int64) ([]json.RawMessage, error)
  • func (r *VKExecutor) ListTrim(ctx context.Context, key Key, start, stop int64) error
  • func (r *VKExecutor) Set(ctx context.Context, key Key, value json.RawMessage) error
  • func (r *VKExecutor) SetAdd(ctx context.Context, key Key, member json.RawMessage) error
  • func (r *VKExecutor) SetMembers(ctx context.Context, key Key) ([]json.RawMessage, error)
  • func (r *VKExecutor) SetRemove(ctx context.Context, key Key, member json.RawMessage) error
  • func (r *VKExecutor) SetWithTTL(ctx context.Context, key Key, value json.RawMessage, ttl time.Duration) error
• type VKManager
  • func NewManager(cfg Config, ttl time.Duration) (*VKManager, error)
  • func (m *VKManager) Close() error
  • func (m *VKManager) Executor(ctx context.Context) (KVExecutor, error)

Constants

const SQLiteSchema = `` /* 194-byte string literal not displayed */

Schema is the DDL needed to bootstrap the kv store table in SQLite. Call this once after opening the database (NewSQLiteDBManager accepts a schema string).

Variables

var (
	ErrNotFound         = errors.New("key not found")
	ErrInvalidValue     = errors.New("invalid value for operation")
	ErrKeyExists        = errors.New("key already exists")
	ErrTxFailed         = errors.New("transaction failed")
	ErrTxConflict       = errors.New("transaction conflict")
	ErrLockUnavailable  = errors.New("lock unavailable")
	ErrConnectionFailed = errors.New("connection failed")
)

Functions

func SetupLocalPDInstance

func SetupLocalPDInstance(ctx context.Context) (string, testcontainers.Container, func(), error)

func SetupLocalTiKVInstance

func SetupLocalTiKVInstance(ctx context.Context) (string, testcontainers.Container, func(), error)

func SetupLocalValKeyInstance

func SetupLocalValKeyInstance(ctx context.Context) (string, testcontainers.Container, func(), error)

Types

type Config

type Config struct {
	KVAddr     string `json:"kv_addr"`
	KVPassword string `json:"kv_password,omitempty"`
}

type KVExecutor

type KVExecutor interface {
	Get(ctx context.Context, key Key) (json.RawMessage, error)
	Set(ctx context.Context, key Key, value json.RawMessage) error
	SetWithTTL(ctx context.Context, key Key, value json.RawMessage, ttl time.Duration) error
	Delete(ctx context.Context, key Key) error
	Exists(ctx context.Context, key Key) (bool, error)
	Keys(ctx context.Context, pattern string) ([]Key, error)

	// List operations
	ListPush(ctx context.Context, key Key, value json.RawMessage) error
	ListRange(ctx context.Context, key Key, start, stop int64) ([]json.RawMessage, error)
	ListTrim(ctx context.Context, key Key, start, stop int64) error
	ListLength(ctx context.Context, key Key) (int64, error)
	ListRPop(ctx context.Context, key Key) (json.RawMessage, error)

	// Set operations
	SetAdd(ctx context.Context, key Key, member json.RawMessage) error
	SetMembers(ctx context.Context, key Key) ([]json.RawMessage, error)
	SetRemove(ctx context.Context, key Key, member json.RawMessage) error
}

KVExecutor is a non-transactional key-value backend.

type KVManager

type KVManager interface {
	Executor(ctx context.Context) (KVExecutor, error)
	Close() error
}

KVManager obtains non-transactional KVExecutors.

type Key

type Key = string

type KeyValue

type KeyValue struct {
	Key   string
	Value json.RawMessage
	TTL   time.Time
}

KeyValue exists for backward compatibility with the old interface.

type SQLiteManager

type SQLiteManager struct {
	// contains filtered or unexported fields
}

SQLiteManager implements KVManager on top of a libdbexec.DBManager (SQLite).

func NewSQLiteManager

func NewSQLiteManager(db libdbexec.DBManager) *SQLiteManager

NewSQLiteManager wraps an existing libdbexec.DBManager. The caller is responsible for opening the database and applying SQLiteSchema.

func (*SQLiteManager) Close

func (m *SQLiteManager) Close() error

Close closes the underlying database.

func (*SQLiteManager) Executor

func (m *SQLiteManager) Executor(_ context.Context) (KVExecutor, error)

Executor returns a KVExecutor bound to a non-transactional connection. The manager is handed along too, so list/set read-modify-write ops can open a real transaction; a bare libdbexec.Exec cannot start one.

type VKExecutor

type VKExecutor struct {
	// contains filtered or unexported fields
}

func (*VKExecutor) Delete

func (r *VKExecutor) Delete(ctx context.Context, key Key) error

func (*VKExecutor) Exists

func (r *VKExecutor) Exists(ctx context.Context, key Key) (bool, error)

func (*VKExecutor) Get

func (r *VKExecutor) Get(ctx context.Context, key Key) (json.RawMessage, error)

func (*VKExecutor) Keys

func (r *VKExecutor) Keys(ctx context.Context, pattern string) ([]Key, error)

func (*VKExecutor) ListLength

func (r *VKExecutor) ListLength(ctx context.Context, key Key) (int64, error)

func (*VKExecutor) ListPush

func (r *VKExecutor) ListPush(ctx context.Context, key Key, value json.RawMessage) error

func (*VKExecutor) ListRPop

func (r *VKExecutor) ListRPop(ctx context.Context, key Key) (json.RawMessage, error)

func (*VKExecutor) ListRange

func (r *VKExecutor) ListRange(ctx context.Context, key Key, start, stop int64) ([]json.RawMessage, error)

func (*VKExecutor) ListTrim

func (r *VKExecutor) ListTrim(ctx context.Context, key Key, start, stop int64) error

func (*VKExecutor) Set

func (r *VKExecutor) Set(ctx context.Context, key Key, value json.RawMessage) error

func (*VKExecutor) SetAdd

func (r *VKExecutor) SetAdd(ctx context.Context, key Key, member json.RawMessage) error

func (*VKExecutor) SetMembers

func (r *VKExecutor) SetMembers(ctx context.Context, key Key) ([]json.RawMessage, error)

func (*VKExecutor) SetRemove

func (r *VKExecutor) SetRemove(ctx context.Context, key Key, member json.RawMessage) error

func (*VKExecutor) SetWithTTL

func (r *VKExecutor) SetWithTTL(ctx context.Context, key Key, value json.RawMessage, ttl time.Duration) error

type VKManager

type VKManager struct {
	// contains filtered or unexported fields
}

func NewManager

func NewManager(cfg Config, ttl time.Duration) (*VKManager, error)

func (*VKManager) Close

func (m *VKManager) Close() error

func (*VKManager) Executor

func (m *VKManager) Executor(ctx context.Context) (KVExecutor, error)