frostfs-node/pkg/ape/chainbase/boltdb.go

package chainbase

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"path/filepath"
	"time"

	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/util/logicerr"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util"
	"git.frostfs.info/TrueCloudLab/policy-engine/pkg/chain"
	policyengine "git.frostfs.info/TrueCloudLab/policy-engine/pkg/engine"
	"github.com/nspcc-dev/neo-go/pkg/util/slice"
	"go.etcd.io/bbolt"
)

type boltLocalOverrideStorage struct {
	*cfg

	db *bbolt.DB
}

var chainBucket = []byte{0}

var (
	// ErrRootBucketNotFound signals the database has not been properly initialized.
	ErrRootBucketNotFound = logicerr.New("root bucket not found")

	ErrGlobalNamespaceBucketNotFound = logicerr.New("global namespace bucket not found")

	ErrTargetTypeBucketNotFound = logicerr.New("target type bucket not found")

	ErrTargetNameBucketNotFound = logicerr.New("target name bucket not found")

	ErrBucketNotContainsChainID = logicerr.New("chain id not found in bucket")
)

// NewBoltLocalOverrideDatabase returns storage wrapper for storing access policy engine
// local overrides.
//
// chain storage (chainBucket):
// -> global namespace bucket (nBucket):
// --> target bucket (tBucket)
// ---> target name (resource) bucket (rBucket):
//
// |        Key          |      Value        |
// x---------------------x-------------------x
// |  chain id (string)  | serialized chain  |
// x---------------------x-------------------x
//
//nolint:godot
func NewBoltLocalOverrideDatabase(opts ...Option) LocalOverrideDatabase {
	c := defaultCfg()

	for i := range opts {
		opts[i](c)
	}

	return &boltLocalOverrideStorage{
		cfg: c,
	}
}

func (cs *boltLocalOverrideStorage) Init() error {
	return cs.db.Update(func(tx *bbolt.Tx) error {
		_, err := tx.CreateBucketIfNotExists(chainBucket)
		return err
	})
}

func (cs *boltLocalOverrideStorage) Open(context.Context) error {
	err := util.MkdirAllX(filepath.Dir(cs.path), cs.perm)
	if err != nil {
		return fmt.Errorf("can't create dir %s for the chain DB: %w", cs.path, err)
	}

	opts := *bbolt.DefaultOptions
	opts.NoSync = cs.noSync
	opts.Timeout = 100 * time.Millisecond

	cs.db, err = bbolt.Open(cs.path, cs.perm, &opts)
	if err != nil {
		return fmt.Errorf("can't open the chain DB: %w", err)
	}

	cs.db.MaxBatchSize = cs.maxBatchSize
	cs.db.MaxBatchDelay = cs.maxBatchDelay

	return nil
}

func (cs *boltLocalOverrideStorage) Close() error {
	var err error
	if cs.db != nil {
		err = cs.db.Close()
	}
	return err
}

func getTargetBucket(tx *bbolt.Tx, name chain.Name, target policyengine.Target) (*bbolt.Bucket, error) {
	cbucket := tx.Bucket(chainBucket)
	if cbucket == nil {
		return nil, ErrRootBucketNotFound
	}

	nbucket := cbucket.Bucket([]byte(name))
	if nbucket == nil {
		return nil, fmt.Errorf("%w: %w: %s", policyengine.ErrChainNotFound, ErrGlobalNamespaceBucketNotFound, name)
	}

	typeBucket := nbucket.Bucket([]byte{byte(target.Type)})
	if typeBucket == nil {
		return nil, fmt.Errorf("%w: %w: %c", policyengine.ErrChainNotFound, ErrTargetTypeBucketNotFound, target.Type)
	}

	rbucket := typeBucket.Bucket([]byte(target.Name))
	if rbucket == nil {
		return nil, fmt.Errorf("%w: %w: %s", policyengine.ErrChainNotFound, ErrTargetNameBucketNotFound, target.Name)
	}
	return rbucket, nil
}

func getTargetBucketCreateIfEmpty(tx *bbolt.Tx, name chain.Name, target policyengine.Target) (*bbolt.Bucket, error) {
	cbucket := tx.Bucket(chainBucket)
	if cbucket == nil {
		return nil, ErrRootBucketNotFound
	}

	nbucket := cbucket.Bucket([]byte(name))
	if nbucket == nil {
		var err error
		nbucket, err = cbucket.CreateBucket([]byte(name))
		if err != nil {
			return nil, fmt.Errorf("could not create a bucket for the global chain name %s: %w", name, err)
		}
	}

	typeBucket := nbucket.Bucket([]byte{byte(target.Type)})
	if typeBucket == nil {
		var err error
		typeBucket, err = nbucket.CreateBucket([]byte{byte(target.Type)})
		if err != nil {
			return nil, fmt.Errorf("could not create a bucket for the target type '%c': %w", target.Type, err)
		}
	}

	rbucket := typeBucket.Bucket([]byte(target.Name))
	if rbucket == nil {
		var err error
		rbucket, err = typeBucket.CreateBucket([]byte(target.Name))
		if err != nil {
			return nil, fmt.Errorf("could not create a bucket for the target name %s: %w", target.Name, err)
		}
	}

	return rbucket, nil
}

func (cs *boltLocalOverrideStorage) AddOverride(name chain.Name, target policyengine.Target, c *chain.Chain) (chain.ID, error) {
	if c.ID == "" {
		return "", fmt.Errorf("chain ID is not set")
	}

	serializedChain := c.Bytes()

	err := cs.db.Update(func(tx *bbolt.Tx) error {
		rbuck, err := getTargetBucketCreateIfEmpty(tx, name, target)
		if err != nil {
			return err
		}
		return rbuck.Put([]byte(c.ID), serializedChain)
	})

	return c.ID, err
}

func (cs *boltLocalOverrideStorage) GetOverride(name chain.Name, target policyengine.Target, chainID chain.ID) (*chain.Chain, error) {
	var serializedChain []byte

	if err := cs.db.View(func(tx *bbolt.Tx) error {
		rbuck, err := getTargetBucket(tx, name, target)
		if err != nil {
			return err
		}
		serializedChain = rbuck.Get([]byte(chainID))
		if serializedChain == nil {
			return fmt.Errorf("%w: %w: %s", policyengine.ErrChainNotFound, ErrBucketNotContainsChainID, chainID)
		}
		serializedChain = slice.Copy(serializedChain)
		return nil
	}); err != nil {
		return nil, err
	}

	c := &chain.Chain{}
	if err := json.Unmarshal(serializedChain, c); err != nil {
		return nil, err
	}
	return c, nil
}

func (cs *boltLocalOverrideStorage) RemoveOverride(name chain.Name, target policyengine.Target, chainID chain.ID) error {
	return cs.db.Update(func(tx *bbolt.Tx) error {
		rbuck, err := getTargetBucket(tx, name, target)
		if err != nil {
			return err
		}
		return rbuck.Delete([]byte(chainID))
	})
}

func (cs *boltLocalOverrideStorage) ListOverrides(name chain.Name, target policyengine.Target) ([]*chain.Chain, error) {
	var serializedChains [][]byte
	var serializedChain []byte
	if err := cs.db.View(func(tx *bbolt.Tx) error {
		rbuck, err := getTargetBucket(tx, name, target)
		if err != nil {
			return err
		}
		return rbuck.ForEach(func(_, v []byte) error {
			serializedChain = slice.Copy(v)
			serializedChains = append(serializedChains, serializedChain)
			return nil
		})
	}); err != nil {
		if errors.Is(err, policyengine.ErrChainNotFound) {
			return []*chain.Chain{}, nil
		}
		return nil, err
	}
	chains := make([]*chain.Chain, 0, len(serializedChains))
	for _, serializedChain = range serializedChains {
		c := &chain.Chain{}
		if err := json.Unmarshal(serializedChain, c); err != nil {
			return nil, err
		}
		chains = append(chains, c)
	}
	return chains, nil
}

func (cs *boltLocalOverrideStorage) DropAllOverrides(name chain.Name) error {
	return cs.db.Update(func(tx *bbolt.Tx) error {
		cbucket := tx.Bucket(chainBucket)
		if cbucket == nil {
			return ErrRootBucketNotFound
		}

		nbucket := cbucket.Bucket([]byte(name))
		if nbucket == nil {
			return fmt.Errorf("%w: %w: global namespace %s", policyengine.ErrChainNotFound, ErrGlobalNamespaceBucketNotFound, name)
		}

		return tx.DeleteBucket([]byte(name))
	})
}