Add a store interface with default implementation

The store interface is a simplified version of client go store. If an
object can be cached there, it should be in our store implementations.

We also provide a default way to generate keys, but we allow users to
provide their own function. This should enable all needed levels of
isolation.

Signed-off-by: Soule BA <bah.soule@gmail.com>

cache/cache.go

@@ -0,0 +1,328 @@
+/*
+Copyright 2024 The Flux authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package cache
+
+import (
+	"fmt"
+	"sync"
+	"time"
+)
+
+// Cache is a thread-safe in-memory key/object store.
+type Cache struct {
+	*cache
+	// keyFunc is used to make the key for objects stored in and retrieved from items, and
+	// should be deterministic.
+	keyFunc KeyFunc
+}
+
+// Item is an item stored in the cache.
+type Item struct {
+	// Object is the item's object.
+	Object any
+	// Expiration is the item's expiration time.
+	Expiration int64
+}
+
+type cache struct {
+	// Items holds the elements in the cache.
+	Items map[string]Item
+	// MaxItems is the maximum number of items the cache can hold.
+	MaxItems int
+	mu       sync.RWMutex
+	janitor  *janitor
+	closed   bool
+}
+
+var _ Expirable = &Cache{}
+
+// New creates a new cache with the given configuration.
+func New(maxItems int, keyFunc KeyFunc, interval time.Duration) *Cache {
+	c := &cache{
+		Items:    make(map[string]Item),
+		MaxItems: maxItems,
+		janitor: &janitor{
+			interval: interval,
+			stop:     make(chan bool),
+		},
+	}
+
+	C := &Cache{cache: c, keyFunc: keyFunc}
+
+	if interval > 0 {
+		go c.janitor.run(c)
+	}
+
+	return C
+}
+
+func (c *Cache) Close() error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.closed {
+		return fmt.Errorf("cache already closed")
+	}
+	c.janitor.stop <- true
+	c.closed = true
+	return nil
+}
+
+// Add an item to the cache, existing items will not be overwritten.
+// To overwrite existing items, use Update.
+// If the cache is full, Add will return an error.
+func (c *Cache) Add(object any) error {
+	key, err := c.keyFunc(object)
+	if err != nil {
+		return KeyError{object, err}
+	}
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.closed {
+		return KeyError{object, fmt.Errorf("cache already closed")}
+	}
+	_, found := c.Items[key]
+	if found {
+		return KeyError{object, fmt.Errorf("key already exists")}
+	}
+
+	if c.MaxItems > 0 && len(c.Items) < c.MaxItems {
+		c.set(key, object)
+		return nil
+	}
+
+	return KeyError{object, fmt.Errorf("Cache is full")}
+}
+
+// Update adds an item to the cache, replacing any existing item.
+// If the cache is full, Set will return an error.
+func (c *Cache) Update(object any) error {
+	key, err := c.keyFunc(object)
+	if err != nil {
+		return KeyError{object, err}
+	}
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.closed {
+		return KeyError{object, fmt.Errorf("cache already closed")}
+	}
+	_, found := c.Items[key]
+	if found {
+		c.set(key, object)
+		return nil
+	}
+
+	if c.MaxItems > 0 && len(c.Items) < c.MaxItems {
+		c.set(key, object)
+		return nil
+	}
+
+	return KeyError{object, fmt.Errorf("Cache is full")}
+}
+
+func (c *cache) set(key string, object any) {
+	var e int64
+	c.Items[key] = Item{
+		Object:     object,
+		Expiration: e,
+	}
+}
+
+// Get an item from the cache. Returns the item or nil, and a bool indicating
+// whether the key was found.
+func (c *Cache) Get(object any) (any, bool, error) {
+	key, err := c.keyFunc(object)
+	if err != nil {
+		return nil, false, KeyError{object, err}
+	}
+	item, found, err := c.get(key)
+	if err != nil {
+		return nil, false, KeyError{object, err}
+	}
+	if !found {
+		return nil, false, nil
+	}
+	return item, true, nil
+}
+
+func (c *Cache) GetByKey(key string) (any, bool, error) {
+	return c.get(key)
+}
+
+func (c *cache) get(key string) (any, bool, error) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	if c.closed {
+		return nil, false, fmt.Errorf("cache already closed")
+	}
+	item, found := c.Items[key]
+	if !found {
+		return nil, false, nil
+	}
+	if item.Expiration > 0 {
+		if item.Expiration < time.Now().UnixNano() {
+			return nil, false, fmt.Errorf("key has expired")
+		}
+	}
+	return item.Object, true, nil
+}
+
+// Delete an item from the cache. Does nothing if the key is not in the cache.
+func (c *Cache) Delete(object any) error {
+	key, err := c.keyFunc(object)
+	if err != nil {
+		return KeyError{object, err}
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.closed {
+		return KeyError{object, fmt.Errorf("cache already closed")}
+	}
+	delete(c.Items, key)
+	return nil
+}
+
+// Clear all items from the cache.
+// This reallocates the underlying array holding the items,
+// so that the memory used by the items is reclaimed.
+// A closed cache cannot be cleared.
+func (c *cache) Clear() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.closed {
+		return
+	}
+	c.Items = make(map[string]Item)
+}
+
+// ListKeys returns a slice of the keys in the cache.
+// If the cache is closed, ListKeys returns nil.
+func (c *cache) ListKeys() []string {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	if c.closed {
+		return nil
+	}
+	keys := make([]string, 0, len(c.Items))
+	for k := range c.Items {
+		keys = append(keys, k)
+	}
+	return keys
+}
+
+// HasExpired returns true if the item has expired.
+func (c *Cache) HasExpired(object any) (bool, error) {
+	key, err := c.keyFunc(object)
+	if err != nil {
+		return false, KeyError{object, err}
+	}
+
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	if c.closed {
+		return false, KeyError{object, fmt.Errorf("cache already closed")}
+	}
+	item, ok := c.Items[key]
+	if !ok {
+		return true, nil
+	}
+	if item.Expiration > 0 {
+		if item.Expiration < time.Now().UnixNano() {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
+// SetExpiration sets the expiration for the given key.
+// Does nothing if the key is not in the cache.
+func (c *Cache) SetExpiration(object any, expiration time.Duration) error {
+	key, err := c.keyFunc(object)
+	if err != nil {
+		return KeyError{object, err}
+	}
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.closed {
+		return KeyError{object, fmt.Errorf("cache already closed")}
+	}
+	item, ok := c.Items[key]
+	if ok {
+		item.Expiration = time.Now().Add(expiration).UnixNano()
+		c.Items[key] = item
+	}
+	return nil
+}
+
+// GetExpiration returns the expiration for the given key.
+// Returns zero if the key is not in the cache or the item
+// has already expired.
+func (c *Cache) GetExpiration(object any) (time.Duration, error) {
+	key, err := c.keyFunc(object)
+	if err != nil {
+		return 0, KeyError{object, err}
+	}
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	if c.closed {
+		return 0, KeyError{object, fmt.Errorf("cache already closed")}
+	}
+	item, ok := c.Items[key]
+	if !ok {
+		return 0, KeyError{object, fmt.Errorf("key not found")}
+	}
+	if item.Expiration > 0 {
+		if item.Expiration < time.Now().UnixNano() {
+			return 0, KeyError{object, fmt.Errorf("key has expired")}
+		}
+	}
+	return time.Duration(item.Expiration - time.Now().UnixNano()), nil
+}
+
+// DeleteExpired deletes all expired items from the cache.
+func (c *cache) DeleteExpired() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.closed {
+		return
+	}
+	for k, v := range c.Items {
+		if v.Expiration > 0 && v.Expiration < time.Now().UnixNano() {
+			delete(c.Items, k)
+		}
+	}
+}
+
+type janitor struct {
+	interval time.Duration
+	stop     chan bool
+}
+
+func (j *janitor) run(c *cache) {
+	ticker := time.NewTicker(j.interval)
+	for {
+		select {
+		case <-ticker.C:
+			c.DeleteExpired()
+		case <-j.stop:
+			ticker.Stop()
+			return
+		}
+	}
+}