edge/pkg/storage/driver/cache/driver.go

package cache

import (
	"bytes"
	"io"
	"net/url"
	"strconv"
	"time"

	"forge.cadoles.com/arcad/edge/pkg/storage"
	"forge.cadoles.com/arcad/edge/pkg/storage/driver"
	"forge.cadoles.com/arcad/edge/pkg/storage/driver/cache/lfu"
	"forge.cadoles.com/arcad/edge/pkg/storage/driver/cache/lfu/fs"
	"forge.cadoles.com/arcad/edge/pkg/storage/driver/cache/lfu/memory"
	"github.com/inhies/go-bytesize"
	"github.com/pkg/errors"
)

func init() {
	driver.RegisterBlobStoreFactory("cache", blobStoreFactory)
}

func blobStoreFactory(dsn *url.URL) (storage.BlobStore, error) {
	query := dsn.Query()

	rawDriver := query.Get("driver")
	if rawDriver == "" {
		return nil, errors.New("missing required url parameter 'driver'")
	}

	query.Del("driver")

	blobStoreOptionFuncs := make([]OptionFunc, 0)

	cacheTTL, err := parseDuration(&query, "cacheTTL")
	if err != nil {
		if !errors.Is(err, errNotFound) {
			return nil, errors.WithStack(err)
		}

		cacheTTL = time.Hour
	}

	blobStoreOptionFuncs = append(blobStoreOptionFuncs, WithCacheTTL(cacheTTL))

	blobBucketCacheSize, err := parseInt(&query, "blobBucketCacheSize")
	if err != nil {
		if !errors.Is(err, errNotFound) {
			return nil, errors.WithStack(err)
		}

		blobBucketCacheSize = 16
	}

	blobStoreOptionFuncs = append(blobStoreOptionFuncs, WithBlobBucketCacheSize(int(blobBucketCacheSize)))

	blobBucketCacheStorePrefix := "blobBucketCacheStore"
	blobBucketCacheStore, err := parseCacheStore[string, storage.BlobBucket](&query, blobBucketCacheStorePrefix)
	if err != nil {
		return nil, errors.WithStack(err)
	}

	blobStoreOptionFuncs = append(blobStoreOptionFuncs, WithBlobBucketCacheStore(blobBucketCacheStore))

	bloInfoCacheSize, err := parseInt(&query, "bloInfoCacheSize")
	if err != nil {
		if !errors.Is(err, errNotFound) {
			return nil, errors.WithStack(err)
		}

		bloInfoCacheSize = 16
	}

	blobStoreOptionFuncs = append(blobStoreOptionFuncs, WithBlobInfoCacheSize(int(bloInfoCacheSize)))

	blobInfoCacheStorePrefix := "blobInfoCacheStore"
	blobInfoCacheStore, err := parseCacheStore[string, storage.BlobInfo](&query, blobInfoCacheStorePrefix)
	if err != nil {
		return nil, errors.WithStack(err)
	}

	blobStoreOptionFuncs = append(blobStoreOptionFuncs, WithBlobInfoCacheStore(blobInfoCacheStore))

	blobCacheSize, err := parseByteSize(&query, "blobCacheSize")
	if err != nil {
		if !errors.Is(err, errNotFound) {
			return nil, errors.WithStack(err)
		}

		blobCacheSize = 256e+6
	}

	blobStoreOptionFuncs = append(blobStoreOptionFuncs, WithBlobCacheSize(int(blobCacheSize)))

	blobCacheStorePrefix := "blobCacheStore"
	blobCacheStore, err := parseCacheStore[string, []byte](
		&query, blobCacheStorePrefix,
		fs.WithMarshalValue[string, []byte](func(value []byte) (io.Reader, error) {
			return bytes.NewBuffer(value), nil
		}),
		fs.WithUnmarshalValue[string, []byte](func(r io.Reader) ([]byte, error) {
			data, err := io.ReadAll(r)
			if err != nil {
				return nil, errors.WithStack(err)
			}

			return data, nil
		}),
	)
	if err != nil {
		return nil, errors.WithStack(err)
	}

	blobStoreOptionFuncs = append(blobStoreOptionFuncs, WithBlobCacheStore(blobCacheStore))

	url := &url.URL{
		Scheme:   rawDriver,
		Host:     dsn.Host,
		Path:     dsn.Path,
		RawQuery: query.Encode(),
	}

	backend, err := driver.NewBlobStore(url.String())
	if err != nil {
		return nil, errors.WithStack(err)
	}

	store, err := NewBlobStore(backend, blobStoreOptionFuncs...)
	if err != nil {
		return nil, errors.WithStack(err)
	}

	return store, nil
}

var errNotFound = errors.New("not found")

func parseString(query *url.Values, name string) (string, error) {
	value := query.Get(name)
	if value != "" {
		query.Del(name)
		return value, nil
	}

	return "", errors.WithStack(errNotFound)
}

func parseByteSize(query *url.Values, name string) (bytesize.ByteSize, error) {
	rawValue := query.Get(name)
	if rawValue != "" {
		query.Del(name)

		value, err := bytesize.Parse(rawValue)
		if err != nil {
			return 0, errors.Wrapf(err, "could not parse url parameter '%s'", name)
		}

		return value, nil
	}

	return 0, errors.WithStack(errNotFound)
}

func parseInt(query *url.Values, name string) (int64, error) {
	rawValue := query.Get(name)
	if rawValue != "" {
		query.Del(name)

		value, err := strconv.ParseInt(rawValue, 10, 32)
		if err != nil {
			return 0, errors.Wrapf(err, "could not parse url parameter '%s'", name)
		}

		return value, nil
	}

	return 0, errors.WithStack(errNotFound)
}

func parseDuration(query *url.Values, name string) (time.Duration, error) {
	rawValue := query.Get(name)
	if rawValue != "" {
		query.Del(name)

		value, err := time.ParseDuration(rawValue)
		if err != nil {
			return 0, errors.Wrapf(err, "could not parse url parameter '%s'", name)
		}

		return value, nil
	}

	return 0, errors.WithStack(errNotFound)
}

const (
	storeTypeFS     string = "fs"
	storeTypeMemory string = "memory"
)

func parseCacheStore[K comparable, V any](query *url.Values, prefix string, optionFuncs ...any) (lfu.Store[K, V], error) {
	storeTypeParam := prefix + "Type"
	storeType, err := parseString(query, storeTypeParam)
	if err != nil {
		if errors.Is(err, errNotFound) {
			storeType = storeTypeMemory
		}
	}

	switch storeType {
	case storeTypeFS:
		store, err := parseFSCacheStore[K, V](query, prefix, optionFuncs...)
		if err != nil {
			return nil, errors.WithStack(err)
		}

		return store, nil

	case storeTypeMemory:
		store, err := parseMemoryCacheStore[K, V](query, prefix, optionFuncs...)
		if err != nil {
			return nil, errors.WithStack(err)
		}

		return store, nil
	}

	return nil, errors.Errorf("unexpected store type value '%s' for parameter '%s'", storeType, storeTypeParam)
}

func parseFSCacheStore[K comparable, V any](query *url.Values, prefix string, optionFuncs ...any) (*fs.Store[K, V], error) {
	baseDirParam := prefix + "BaseDir"
	baseDir, err := parseString(query, baseDirParam)
	if err != nil {
		if errors.Is(err, errNotFound) {
			return nil, errors.Wrapf(err, "missing required url parameter '%s'", baseDirParam)
		}

		return nil, errors.WithStack(err)
	}

	funcs := make([]fs.OptionsFunc[K, V], 0)

	for _, anyFn := range optionFuncs {
		fn, ok := anyFn.(fs.OptionsFunc[K, V])
		if !ok {
			continue
		}

		funcs = append(funcs, fn)
	}

	store := fs.NewStore[K, V](baseDir, funcs...)

	if err := store.Clear(); err != nil {
		return nil, errors.WithStack(err)
	}

	return store, nil
}

func parseMemoryCacheStore[K comparable, V any](query *url.Values, prefix string, optionFuncs ...any) (*memory.Store[K, V], error) {
	return memory.NewStore[K, V](), nil
}