Skip to content

Commit

Permalink
feat(ARCO-283): split cache methods
Browse files Browse the repository at this point in the history
  • Loading branch information
@pawellewandowski98
pawellewandowski98 committed Nov 20, 2024
1 parent 8093791 commit f43cf5a
Show file tree
Hide file tree
Showing 4 changed files with 127 additions and 102 deletions.
16 changes: 10 additions & 6 deletions internal/cache/cache.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,10 +16,14 @@ var (
)

type Store interface {
Get(hash *string, key string) ([]byte, error)
GetAllForHash(hash string) (map[string][]byte, error)
GetAllForHashAndDelete(hash string) (map[string][]byte, error)
Set(hash *string, key string, value []byte, ttl time.Duration) error
Del(hash *string, keys ...string) error
CountElementsForHash(hash string) (int64, error)
Get(key string) ([]byte, error)
Set(key string, value []byte, ttl time.Duration) error
Del(keys ...string) error

MapGet(hash string, key string) ([]byte, error)
MapGetAll(hash string) (map[string][]byte, error)
MapSet(hash string, key string, value []byte) error
MapDel(hash string, keys ...string) error
MapLen(hash string) (int64, error)
MapExtractAll(hash string) (map[string][]byte, error)
}
121 changes: 62 additions & 59 deletions internal/cache/in_memory.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,27 +14,8 @@ func NewMemoryStore() *MemoryStore {
return &MemoryStore{}
}

// Get retrieves a value by key and hash (if given)
func (s *MemoryStore) Get(hash *string, key string) ([]byte, error) {
if hash != nil {
hashValue, found := s.data.Load(hash)
if !found {
return nil, ErrCacheNotFound
}

hashMap, ok := hashValue.(map[string][]byte)
if !ok {
return nil, ErrCacheFailedToGet
}

fieldValue, exists := hashMap[key]
if !exists {
return nil, ErrCacheNotFound
}

return fieldValue, nil
}

// Get retrieves a value by key.
func (s *MemoryStore) Get(key string) ([]byte, error) {
value, found := s.data.Load(key)
if !found {
return nil, ErrCacheNotFound
Expand All @@ -49,54 +30,76 @@ func (s *MemoryStore) Get(hash *string, key string) ([]byte, error) {
}

// Set stores a key-value pair, ignoring the ttl parameter.
func (s *MemoryStore) Set(hash *string, key string, value []byte, _ time.Duration) error {
if hash != nil {
raw, _ := s.data.LoadOrStore(*hash, make(map[string][]byte))
func (s *MemoryStore) Set(key string, value []byte, _ time.Duration) error {
s.data.Store(key, value)
return nil
}

hashMap, ok := raw.(map[string][]byte)
if !ok {
return ErrCacheFailedToSet
}
// Del removes a key from the store.
func (s *MemoryStore) Del(keys ...string) error {
for _, k := range keys {
s.data.Delete(k)
}
return nil
}

hashMap[key] = value
// MapGet retrieves a value by key and hash.
func (s *MemoryStore) MapGet(hash string, key string) ([]byte, error) {
hashValue, found := s.data.Load(hash)
if !found {
return nil, ErrCacheNotFound
}

s.data.Store(*hash, hashMap)
return nil
hashMap, ok := hashValue.(map[string][]byte)
if !ok {
return nil, ErrCacheFailedToGet
}

s.data.Store(key, value)
return nil
fieldValue, exists := hashMap[key]
if !exists {
return nil, ErrCacheNotFound
}

return fieldValue, nil
}

// Del removes a key from the store.
func (s *MemoryStore) Del(hash *string, keys ...string) error {
if hash != nil {
hashValue, found := s.data.Load(*hash)
if !found {
return ErrCacheNotFound
}
// MapSet stores a key-value pair for specific hash.
func (s *MemoryStore) MapSet(hash string, key string, value []byte) error {
raw, _ := s.data.LoadOrStore(hash, make(map[string][]byte))

hashMap, ok := hashValue.(map[string][]byte)
if !ok {
return errors.Join(ErrCacheFailedToDel, ErrCacheFailedToGet)
}
hashMap, ok := raw.(map[string][]byte)
if !ok {
return ErrCacheFailedToSet
}

for _, k := range keys {
delete(hashMap, k)
}
hashMap[key] = value

s.data.Store(*hash, hashMap)
return nil
s.data.Store(hash, hashMap)
return nil
}

// MapDel removes a value by key in specific hash.
func (s *MemoryStore) MapDel(hash string, keys ...string) error {
hashValue, found := s.data.Load(hash)
if !found {
return ErrCacheNotFound
}

hashMap, ok := hashValue.(map[string][]byte)
if !ok {
return errors.Join(ErrCacheFailedToDel, ErrCacheFailedToGet)
}

for _, k := range keys {
s.data.Delete(k)
delete(hashMap, k)
}

s.data.Store(hash, hashMap)
return nil
}

// GetAllForHash retrieves all key-value pairs for a specific hash.
func (s *MemoryStore) GetAllForHash(hash string) (map[string][]byte, error) {
// MapGetAll retrieves all key-value pairs for a specific hash.
func (s *MemoryStore) MapGetAll(hash string) (map[string][]byte, error) {
hashValue, found := s.data.Load(hash)
if !found {
return nil, ErrCacheNotFound
Expand All @@ -110,24 +113,24 @@ func (s *MemoryStore) GetAllForHash(hash string) (map[string][]byte, error) {
return hashMap, nil
}

// GetAllForHashAndDelete retrieves all key-value pairs for a specific hash and deletes the hash.
func (s *MemoryStore) GetAllForHashAndDelete(hash string) (map[string][]byte, error) {
hashMap, err := s.GetAllForHash(hash)
// MapExtractAll retrieves all key-value pairs for a specific hash and deletes the hash.
func (s *MemoryStore) MapExtractAll(hash string) (map[string][]byte, error) {
hashMap, err := s.MapGetAll(hash)
if err != nil {
return nil, err
}

err = s.Del(nil, hash)
err = s.Del(hash)
if err != nil {
return nil, err
}

return hashMap, nil
}

// CountElementsForHash returns the number of elements in a hash in memory.
func (s *MemoryStore) CountElementsForHash(hash string) (int64, error) {
hashMap, err := s.GetAllForHash(hash)
// MapLen returns the number of elements in a hash in memory.
func (s *MemoryStore) MapLen(hash string) (int64, error) {
hashMap, err := s.MapGetAll(hash)
if err != nil {
if errors.Is(err, ErrCacheNotFound) {
return 0, nil
Expand Down
84 changes: 51 additions & 33 deletions internal/cache/redis.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,16 +22,9 @@ func NewRedisStore(ctx context.Context, c redis.UniversalClient) *RedisStore {
}
}

// Get retrieves a value by key and hash (if given).
func (r *RedisStore) Get(hash *string, key string) ([]byte, error) {
var result string
var err error

if hash == nil {
result, err = r.client.Get(r.ctx, key).Result()
} else {
result, err = r.client.HGet(r.ctx, *hash, key).Result()
}
// Get retrieves a value by key.
func (r *RedisStore) Get(key string) ([]byte, error) {
result, err := r.client.Get(r.ctx, key).Result()

if errors.Is(err, redis.Nil) {
return nil, ErrCacheNotFound
Expand All @@ -42,15 +35,9 @@ func (r *RedisStore) Get(hash *string, key string) ([]byte, error) {
return []byte(result), nil
}

// Set stores a value with a TTL for a specific hash (if given).
func (r *RedisStore) Set(hash *string, key string, value []byte, ttl time.Duration) error {
var err error

if hash == nil {
err = r.client.Set(r.ctx, key, value, ttl).Err()
} else {
err = r.client.HSet(r.ctx, *hash, key, value).Err()
}
// Set stores a value with a TTL for key.
func (r *RedisStore) Set(key string, value []byte, ttl time.Duration) error {
err := r.client.Set(r.ctx, key, value, ttl).Err()

if err != nil {
return errors.Join(ErrCacheFailedToSet, err)
Expand All @@ -60,27 +47,58 @@ func (r *RedisStore) Set(hash *string, key string, value []byte, ttl time.Durati
}

// Del removes a value by key.
func (r *RedisStore) Del(hash *string, keys ...string) error {
var result int64
var err error

if hash == nil {
result, err = r.client.Del(r.ctx, keys...).Result()
} else {
result, err = r.client.HDel(r.ctx, *hash, keys...).Result()
func (r *RedisStore) Del(keys ...string) error {
result, err := r.client.Del(r.ctx, keys...).Result()

if err != nil {
return errors.Join(ErrCacheFailedToDel, err)
}
if result == 0 {
return ErrCacheNotFound
}
return nil
}

// MapGet retrieves a value by key and hash (if given).
func (r *RedisStore) MapGet(hash string, key string) ([]byte, error) {
result, err := r.client.HGet(r.ctx, hash, key).Result()

if errors.Is(err, redis.Nil) {
return nil, ErrCacheNotFound
} else if err != nil {
return nil, errors.Join(ErrCacheFailedToGet, err)
}

return []byte(result), nil
}

// MapSet stores a value for a specific hash.
func (r *RedisStore) MapSet(hash string, key string, value []byte) error {
err := r.client.HSet(r.ctx, hash, key, value).Err()

if err != nil {
return errors.Join(ErrCacheFailedToSet, err)
}

return nil
}

// MapDel removes a value by key in specific hash.
func (r *RedisStore) MapDel(hash string, keys ...string) error {
result, err := r.client.HDel(r.ctx, hash, keys...).Result()

if err != nil {
return errors.Join(ErrCacheFailedToDel, err)
}
if result == 0 {
return ErrCacheNotFound
}

return nil
}

// GetAllForHash retrieves all key-value pairs for a specific hash.
func (r *RedisStore) GetAllForHash(hash string) (map[string][]byte, error) {
// MapGetAll retrieves all key-value pairs for a specific hash.
func (r *RedisStore) MapGetAll(hash string) (map[string][]byte, error) {
values, err := r.client.HGetAll(r.ctx, hash).Result()
if err != nil {
return nil, errors.Join(ErrCacheFailedToGet, err)
Expand All @@ -93,8 +111,8 @@ func (r *RedisStore) GetAllForHash(hash string) (map[string][]byte, error) {
return result, nil
}

// GetAllForHashAndDelete retrieves all key-value pairs for a specific hash and remove them from cache, all in one transaction.
func (r *RedisStore) GetAllForHashAndDelete(hash string) (map[string][]byte, error) {
// MapExtractAll retrieves all key-value pairs for a specific hash and remove them from cache, all in one transaction.
func (r *RedisStore) MapExtractAll(hash string) (map[string][]byte, error) {
tx := r.client.TxPipeline()

getAllCmd := tx.HGetAll(r.ctx, hash)
Expand All @@ -114,8 +132,8 @@ func (r *RedisStore) GetAllForHashAndDelete(hash string) (map[string][]byte, err
return result, nil
}

// CountElementsForHash returns the number of elements in a hash.
func (r *RedisStore) CountElementsForHash(hash string) (int64, error) {
// MapLen returns the number of elements in a hash.
func (r *RedisStore) MapLen(hash string) (int64, error) {
count, err := r.client.HLen(r.ctx, hash).Result()
if err != nil {
return 0, errors.Join(ErrCacheFailedToGetCount, err)
Expand Down
8 changes: 4 additions & 4 deletions internal/metamorph/processor_helpers.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -49,15 +49,15 @@ func (p *Processor) setTransactionStatus(status store.UpdateStatus) error {
return errors.Join(ErrFailedToSerialize, err)
}

err = p.cacheStore.Set(&CacheStatusUpdateHash, status.Hash.String(), bytes, processStatusUpdatesIntervalDefault)
err = p.cacheStore.MapSet(CacheStatusUpdateHash, status.Hash.String(), bytes)
if err != nil {
return err
}
return nil
}

func (p *Processor) getTransactionStatus(hash chainhash.Hash) (*store.UpdateStatus, error) {
bytes, err := p.cacheStore.Get(&CacheStatusUpdateHash, hash.String())
bytes, err := p.cacheStore.MapGet(CacheStatusUpdateHash, hash.String())
if err != nil {
return nil, err
}
Expand All @@ -73,7 +73,7 @@ func (p *Processor) getTransactionStatus(hash chainhash.Hash) (*store.UpdateStat

func (p *Processor) getAndDeleteAllTransactionStatuses() (StatusUpdateMap, error) {
statuses := make(StatusUpdateMap)
keys, err := p.cacheStore.GetAllForHashAndDelete(CacheStatusUpdateHash)
keys, err := p.cacheStore.MapExtractAll(CacheStatusUpdateHash)
if err != nil {
return nil, err
}
Expand All @@ -97,7 +97,7 @@ func (p *Processor) getAndDeleteAllTransactionStatuses() (StatusUpdateMap, error
}

func (p *Processor) getStatusUpdateCount() (int, error) {
count, err := p.cacheStore.CountElementsForHash(CacheStatusUpdateHash)
count, err := p.cacheStore.MapLen(CacheStatusUpdateHash)
if err != nil {
return 0, err
}
Expand Down

0 comments on commit f43cf5a

Please sign in to comment.