kern/v4/lev2/lsi/kstore_kv/kstore_kv.go

// package kstore_kv -- локальное быстрое key-value хранилище ядра.
package kstore_kv

import (
	"os"
	"sync"
	"time"

	"github.com/dgraph-io/badger/v4"

	mL0 "gitp78su.ipnodns.ru/svi/kern/v4/lev0"
	"gitp78su.ipnodns.ru/svi/kern/v4/lev0/core_spec"
	"gitp78su.ipnodns.ru/svi/kern/v4/lev0/defs"
	"gitp78su.ipnodns.ru/svi/kern/v4/lev0/defs/stream_name"
	"gitp78su.ipnodns.ru/svi/kern/v4/lev0/etypes/ebool"
	mKh "gitp78su.ipnodns.ru/svi/kern/v4/lev0/helpers"
	mL1 "gitp78su.ipnodns.ru/svi/kern/v4/lev1"
	"gitp78su.ipnodns.ru/svi/kern/v4/lev1/comp_spec"
	"gitp78su.ipnodns.ru/svi/kern/v4/lev1/log_buf"
	"gitp78su.ipnodns.ru/svi/kern/v4/lev1/safe_bool"
	"gitp78su.ipnodns.ru/svi/kern/v4/lev2/kctx"
	"gitp78su.ipnodns.ru/svi/kern/v4/lev2/kspec"
	"gitp78su.ipnodns.ru/svi/kern/v4/lev2/lsi/store_ent"
	"gitp78su.ipnodns.ru/svi/kern/v4/lev2/lsi/store_ent/rec_key"
	"gitp78su.ipnodns.ru/svi/kern/v4/lev2/lsi/store_spec"
)

var (
	storeStreamName = stream_name.NewAStreamName(defs.Txt("kstore_kv")) // Имя потока для ожидателя потоков
)

// kStoreKv -- локальное хранилище ядра.
type kStoreKv struct {
	sync.RWMutex
	kCtx      kspec.IKernelCtx
	lCtx      comp_spec.ILocalCtx
	log       comp_spec.ILogBuf
	wg        kspec.IKernelWg
	storePath defs.ITxtFix
	db        *badger.DB
	isWork    core_spec.ISafeBool
}

var (
	kernStore *kStoreKv // Глобальный объект
	block     sync.Mutex
)

// GetKernelStore -- возвращает новое локальное хранилище ядра.
func GetKernelStore() store_spec.IKernelStoreKv {
	block.Lock()
	defer block.Unlock()
	if kernStore != nil {
		kernStore.log.Debug(defs.Txt("GetKernelStore()"))
		return kernStore
	}
	kCtx := kctx.GetKernelCtx()
	param := &log_buf.LogBufParam{
		Prefix_: defs.Txt("kStoreKv"),
		IsTerm_: safe_bool.NewSafeBool(true),
	}
	log := mL1.NewLogBuf(param)
	sf := &kStoreKv{
		kCtx:   kCtx,
		lCtx:   mL1.NewLocalCtx(kCtx.Ctx()),
		wg:     kCtx.Wg(),
		isWork: mL1.NewSafeBool(true),
		log:    log,
	}
	sf.open()
	kernStore = sf
	kCtx.Set(defs.Txt("kernStoreKV"), kernStore, "fast KV store on Badger")
	return kernStore
}

// Log -- возвращает локальный лог.
func (sf *kStoreKv) Log() comp_spec.ILogBuf {
	return sf.log
}

// Set -- устанавливает значение по ключу.
func (sf *kStoreKv) Set(rec store_ent.IRecKvFix) mL0.IResult[core_spec.EBool] {
	sf.Lock()
	defer sf.Unlock()
	key := rec.Key()
	sf.log.Debug(defs.Txt("Set(): key='%v'", key.Get()))
	fnSet := func(txn *badger.Txn) error {
		err := txn.Set(key.Byte(), rec.Val().Byte())
		return err
	}
	err := sf.db.Update(fnSet)
	if err != nil {
		err := defs.Err("Set(): key=%v, err=\n\t%w", key, err)
		sf.log.Err(err)
		return mL0.NewErr[core_spec.EBool](err)
	}
	return mL0.NewOk(ebool.NewEBool(true))
}

// Get -- возвращает значение по ключу.
func (sf *kStoreKv) Get(key store_ent.IStoreKeyFix) mL0.IResult[[]byte] {
	sf.RLock()
	defer sf.RUnlock()
	sf.log.Debug(defs.Txt("Get(): key='%v'", key))
	var binVal []byte
	fnGet := func(txn *badger.Txn) error {
		item, err := txn.Get(key.Byte())
		if err != nil {
			return err
		}
		binVal, err = item.ValueCopy(binVal)
		return err
	}
	err := sf.db.View(fnGet)
	if err != nil {
		err := defs.Err("Get(): key=%v, err=\n\t%v", key, err)
		sf.log.Err(err)
		return mL0.NewErr[[]byte](err)
	}
	return mL0.NewOk(binVal)
}

// ByPrefix -- фильтрует ключи по префиксу.
func (sf *kStoreKv) ByPrefix(prefix store_ent.IStoreKeyFix) core_spec.IResult[[]store_ent.AStoreKey] {
	var (
		binKey []byte
		lstKey = []store_ent.AStoreKey{}
	)
	// fnValue := func(v []byte) error {
	// 	fmt.Printf("key=%s, value=%s\n", key, v)
	// 	return nil
	// }
	fnPrefix := func(txn *badger.Txn) error {
		it := txn.NewIterator(badger.DefaultIteratorOptions)
		defer it.Close()
		binPref := prefix.Byte()
		for it.Seek(binPref); it.ValidForPrefix(binPref); it.Next() {
			item := it.Item()
			binKey = item.Key()
			// err := item.Value(fnValue)
			// if err != nil {
			// 	return err
			// }
			lstKey = append(lstKey, rec_key.ARecKey(binKey))
		}
		return nil
	}
	err := sf.db.View(fnPrefix)
	if err != nil {
		err := defs.Err("ByPrefix(): in find, err=\n\t%w", err)
		return mL0.NewErr[[]store_ent.AStoreKey](err)
	}
	return mL0.NewOk(lstKey)
}

// Delete -- удалить ключ из хранилища.
func (sf *kStoreKv) Delete(key store_ent.IStoreKeyFix) mL0.IResult[core_spec.EBool] {
	sf.Lock()
	defer sf.Unlock()
	sf.log.Debug(defs.Txt("Delete(): key='%v'", key))
	fnDelete := func(txn *badger.Txn) error {
		err := txn.Delete(key.Byte())
		return err
	}
	err := sf.db.Update(fnDelete)
	if err != nil {
		err := defs.Err("Delete(): key=%v, err=\n\t%w", key, err)
		sf.log.Err(err)
		return mL0.NewErr[core_spec.EBool](err)
	}
	return mL0.NewOk(ebool.NewEBool(true))
}

// Открывает базу при создании.
func (sf *kStoreKv) open() {
	sf.Lock()
	defer sf.Unlock()
	sf.log.Debug(defs.Txt("open()"))
	strPath := os.Getenv("LOCAL_STORE_PATH")
	mKh.Hassert(strPath != "", "open(): env LOCAL_STORE_PATH not set")
	pwd, err := os.Getwd()
	mKh.Hassert(err == nil, "open(): in get PWD, err=\n\t%v", err)
	sf.storePath = defs.Txt(pwd + strPath + "/db_local")
	err = os.MkdirAll(sf.storePath.String(), 0750)
	mKh.Hassert(err == nil, "open(): in make dir %v, err=\n\t%v", sf.storePath, err)
	sf.db, err = badger.Open(badger.DefaultOptions(sf.storePath.String()))
	mKh.Hassert(err == nil, "open(): in open DB %v, err=\n\t%v", sf.storePath, err)
	sf.wg.Add(storeStreamName)
	sf.isWork.Set()
	go sf.close()
	go sf.clean()
}

// Выполняет периодическую сборку мусора в файле.
func (sf *kStoreKv) clean() {
	chRun := make(chan int, 2)
	defer close(chRun)
	fnClean := func() {
		sf.Lock()
		defer sf.Unlock()
		_ = sf.db.RunValueLogGC(0.7)
	}
	chRun <- 1
	for {
		select {
		case <-sf.kCtx.Ctx().Done(): // надо прекратить работу
			return
		case <-chRun: // Пора поработать
			fnClean()
		}
		time.Sleep(time.Second * 1)
	}
}

// Ожидает последнего потока под отдельной блокировкой.
func (sf *kStoreKv) wait(chWait chan int) {
	for {
		time.Sleep(time.Millisecond * 5)
		if sf.wg.Len() <= 1 {
			break
		}
	}
	close(chWait)
}

// Ожидает закрытия контекста ядра, закрывает хранилище.
func (sf *kStoreKv) close() {
	sf.kCtx.Wait()
	sf.Lock()
	defer sf.Unlock()
	if !sf.isWork.Get() {
		return
	}
	chWait := make(chan int, 2)
	go sf.wait(chWait)
	<-chWait
	sf.isWork.Reset()
	err := sf.db.Close()
	mKh.Assert(err == nil, "kStoreKv.close(): in close DB, err=\n\t%v", err)
	sf.wg.Done(storeStreamName)
	sf.log.Debug(defs.Txt("close(): done"))
}