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

package cache

import (
	"context"
	"fmt"
	"io"

	"forge.cadoles.com/arcad/edge/pkg/storage"
	"forge.cadoles.com/arcad/edge/pkg/storage/driver/cache/lfu"
	"github.com/pkg/errors"
	"gitlab.com/wpetit/goweb/logger"
)

type BlobBucket struct {
	bucket        storage.BlobBucket
	blobCache     *lfu.Cache[string, []byte]
	bucketCache   *lfu.Cache[string, storage.BlobBucket]
	blobInfoCache *lfu.Cache[string, storage.BlobInfo]
}

// Close implements storage.BlobBucket.
func (b *BlobBucket) Close() error {
	// Close only when bucket is evicted from cache
	return nil
}

// Delete implements storage.BlobBucket.
func (b *BlobBucket) Delete(ctx context.Context, id storage.BlobID) error {
	defer b.clearCache(ctx, id)

	if err := b.bucket.Delete(ctx, id); err != nil {
		return errors.WithStack(err)
	}

	return nil
}

// Get implements storage.BlobBucket.
func (b *BlobBucket) Get(ctx context.Context, id storage.BlobID) (storage.BlobInfo, error) {
	key := b.getCacheKey(id)
	blobInfo, err := b.blobInfoCache.Get(key)
	if err != nil && !errors.Is(err, lfu.ErrNotFound) {
		logger.Error(
			ctx, "could not retrieve blob info from cache",
			logger.F("cacheKey", key),
			logger.CapturedE(errors.WithStack(err)),
		)
	}

	if blobInfo != nil {
		logger.Debug(
			ctx, "found blob info in cache",
			logger.F("cacheKey", key),
		)

		return blobInfo, nil
	}

	info, err := b.bucket.Get(ctx, id)
	if err != nil {
		if errors.Is(err, storage.ErrBucketClosed) {
			b.clearCache(ctx, id)
			if err := b.bucketCache.Delete(b.Name()); err != nil && !errors.Is(err, lfu.ErrNotFound) {
				logger.Error(
					ctx, "could not delete bucket from cache",
					logger.F("cacheKey", b.Name()),
					logger.CapturedE(errors.WithStack(err)),
				)
			}
		}

		return nil, errors.WithStack(err)
	}

	if err := b.blobInfoCache.Set(key, info); err != nil {
		logger.Error(
			ctx, "could not set blob info in cache",
			logger.F("cacheKey", key),
			logger.CapturedE(errors.WithStack(err)),
		)
	}

	return info, nil
}

// List implements storage.BlobBucket.
func (b *BlobBucket) List(ctx context.Context) ([]storage.BlobInfo, error) {
	infos, err := b.bucket.List(ctx)
	if err != nil {
		if errors.Is(err, storage.ErrBucketClosed) {
			if err := b.bucketCache.Delete(b.Name()); err != nil && !errors.Is(err, lfu.ErrNotFound) {
				logger.Error(
					ctx, "could not delete bucket from cache",
					logger.F("cacheKey", b.Name()),
					logger.CapturedE(errors.WithStack(err)),
				)
			}
		}

		return nil, errors.WithStack(err)
	}

	for _, ifo := range infos {
		key := b.getCacheKey(ifo.ID())
		if err := b.blobInfoCache.Set(key, ifo); err != nil {
			logger.Error(
				ctx, "could not set blob info in cache",
				logger.F("cacheKey", key),
				logger.CapturedE(errors.WithStack(err)),
			)
		}
	}

	return infos, nil
}

// Name implements storage.BlobBucket.
func (b *BlobBucket) Name() string {
	return b.bucket.Name()
}

// NewReader implements storage.BlobBucket.
func (b *BlobBucket) NewReader(ctx context.Context, id storage.BlobID) (io.ReadSeekCloser, error) {
	if cached, exist := b.inContentCache(id); exist {
		logger.Debug(
			ctx, "found blob content in cache",
			logger.F("cacheKey", b.getCacheKey(id)),
		)
		return cached, nil
	}

	reader, err := b.bucket.NewReader(ctx, id)
	if err != nil {
		if errors.Is(err, storage.ErrBucketClosed) {
			b.clearCache(ctx, id)
			if err := b.bucketCache.Delete(b.Name()); err != nil && !errors.Is(err, lfu.ErrNotFound) {
				logger.Error(
					ctx, "could not delete bucket from cache",
					logger.F("cacheKey", b.Name()),
					logger.CapturedE(errors.WithStack(err)),
				)
			}
		}

		return nil, errors.WithStack(err)
	}

	return &readCacher{
		reader: reader,
		cache:  b.blobCache,
		key:    b.getCacheKey(id),
	}, nil
}

func (b *BlobBucket) getCacheKey(id storage.BlobID) string {
	return fmt.Sprintf("%s-%s", b.Name(), id)
}

func (b *BlobBucket) inContentCache(id storage.BlobID) (io.ReadSeekCloser, bool) {
	key := b.getCacheKey(id)
	data, err := b.blobCache.Get(key)
	if err != nil {
		if errors.Is(err, lfu.ErrNotFound) {
			return nil, false
		}

		logger.Error(context.Background(), "could not retrieve cached value", logger.CapturedE(errors.WithStack(err)))

		return nil, false
	}

	return &cachedReader{data, 0}, true
}

func (b *BlobBucket) clearCache(ctx context.Context, id storage.BlobID) {
	key := b.getCacheKey(id)

	logger.Debug(ctx, "clearing cache", logger.F("cacheKey", key))

	if err := b.blobCache.Delete(key); err != nil && !errors.Is(err, lfu.ErrNotFound) {
		logger.Error(ctx, "could not clear cache", logger.F("cacheKey", key), logger.CapturedE(errors.WithStack(err)))
	}

	if err := b.blobInfoCache.Delete(key); err != nil {
		logger.Error(
			ctx, "could not delete blob info from cache",
			logger.F("cacheKey", key),
			logger.CapturedE(errors.WithStack(err)),
		)
	}
}

// NewWriter implements storage.BlobBucket.
func (b *BlobBucket) NewWriter(ctx context.Context, id storage.BlobID) (io.WriteCloser, error) {
	defer b.clearCache(ctx, id)

	writer, err := b.bucket.NewWriter(ctx, id)
	if err != nil {
		if errors.Is(err, storage.ErrBucketClosed) {
			if err := b.bucketCache.Delete(b.Name()); err != nil && !errors.Is(err, lfu.ErrNotFound) {
				logger.Error(
					ctx, "could not delete bucket from cache",
					logger.F("cacheKey", b.Name()),
					logger.CapturedE(errors.WithStack(err)),
				)
			}
		}

		return nil, errors.WithStack(err)
	}

	return writer, nil
}

// Size implements storage.BlobBucket.
func (b *BlobBucket) Size(ctx context.Context) (int64, error) {
	size, err := b.bucket.Size(ctx)
	if err != nil {
		if errors.Is(err, storage.ErrBucketClosed) {
			if err := b.bucketCache.Delete(b.Name()); err != nil && !errors.Is(err, lfu.ErrNotFound) {
				logger.Error(
					ctx, "could not delete bucket from cache",
					logger.F("cacheKey", b.Name()),
					logger.CapturedE(errors.WithStack(err)),
				)
			}
		}

		return 0, errors.WithStack(err)
	}

	return size, nil
}

var _ storage.BlobBucket = &BlobBucket{}