[Streams-adapters]: Cache, helpers, types (#4318)

* cache, helpers, types

* configuration management

* wip address PR comments

* refactored aliases

* address more PR comments

* updated helpers

Author: denis-chernov-smartcontract

packages/streams-adapter/cache/cache.go

@@ -0,0 +1,151 @@
+package cache
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	types "streams-adapter/common"
+	helpers "streams-adapter/helpers"
+)
+
+var _ types.Cache = (*Cache)(nil)
+
+// Config holds cache configuration
+type Config struct {
+	MaxSize         uint          // Maximum number of items
+	TTL             time.Duration // Time-to-live for items
+	CleanupInterval time.Duration // How often to run cleanup
+}
+
+// Cache represents an in-memory cache for observation data
+type Cache struct {
+	mu              sync.RWMutex
+	items           map[string]*types.CacheItem
+	maxSize         uint
+	ttl             time.Duration
+	cleanupInterval time.Duration
+	ctx             context.Context
+	cancel          context.CancelFunc
+	stopOnce        sync.Once
+}
+
+// New creates a new cache instance
+func New(cfg Config) *Cache {
+	ctx, cancel := context.WithCancel(context.Background())
+
+	c := &Cache{
+		items:           make(map[string]*types.CacheItem),
+		maxSize:         cfg.MaxSize,
+		ttl:             cfg.TTL,
+		cleanupInterval: cfg.CleanupInterval,
+		ctx:             ctx,
+		cancel:          cancel,
+	}
+
+	// Start cleanup goroutine
+	go c.cleanupLoop()
+
+	return c
+
+}
+
+// Set stores an observation for the given request parameters with a timestamp
+func (c *Cache) Set(params types.RequestParams, obs *types.Observation, timestamp time.Time, originalAdapterKey string) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	key, err := helpers.CalculateCacheKey(params)
+	if err != nil {
+		// If we cannot generate a cache key, skip caching this observation
+		return
+	}
+	c.items[key] = &types.CacheItem{
+		Observation:        obs,
+		Timestamp:          timestamp,
+		OriginalAdapterKey: originalAdapterKey,
+	}
+}
+
+// Get retrieves an observation for the given request parameters
+func (c *Cache) Get(params types.RequestParams) *types.Observation {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	key, err := helpers.CalculateCacheKey(params)
+	if err != nil {
+		return nil
+	}
+	if item, exists := c.items[key]; exists {
+		return item.Observation
+	}
+	return nil
+}
+
+// Keys returns all keys in the cache
+func (c *Cache) Keys() []string {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	keys := make([]string, 0, len(c.items))
+	for key := range c.items {
+		keys = append(keys, key)
+	}
+	return keys
+}
+
+// Size returns the number of items in the cache
+func (c *Cache) Size() int {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	return len(c.items)
+}
+
+// Items returns all items in the cache
+func (c *Cache) Items() map[string]*types.CacheItem {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	// Create a copy to avoid race conditions
+	items := make(map[string]*types.CacheItem, len(c.items))
+	for key, item := range c.items {
+		items[key] = item
+	}
+	return items
+}
+
+// cleanupLoop periodically removes expired items
+func (c *Cache) cleanupLoop() {
+	ticker := time.NewTicker(c.cleanupInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-c.ctx.Done():
+			return
+		case <-ticker.C:
+			c.cleanupExpired()
+		}
+	}
+}
+
+// cleanupExpired removes expired items from the cache
+func (c *Cache) cleanupExpired() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	cutoff := time.Now().Add(-c.ttl)
+	for key, item := range c.items {
+		if item.Timestamp.Before(cutoff) {
+			delete(c.items, key)
+		}
+	}
+}
+
+// Stop stops the cache cleanup goroutine
+func (c *Cache) Stop() {
+	c.stopOnce.Do(func() {
+		c.cancel()
+	})
+}

packages/streams-adapter/common/types.go

@@ -0,0 +1,43 @@
+package types
+
+import (
+	"time"
+
+	"github.com/goccy/go-json"
+)
+
+// RequestParams represents dynamic request parameters
+// All request parameters are treated equally (endpoint, base, quote, isin, market, fundId, etc.)
+type RequestParams map[string]string
+
+// Observation represents the data returned from an adapter
+type Observation struct {
+	Data    json.RawMessage `json:"data"`
+	Success bool            `json:"success"`
+	Error   string          `json:"error,omitempty"`
+}
+
+// CacheItem represents a cached value with metadata
+type CacheItem struct {
+	Observation        *Observation
+	Timestamp          time.Time // Last write time
+	OriginalAdapterKey string
+}
+
+// Cache interface for storing and retrieving observations
+type Cache interface {
+	Set(params RequestParams, observation *Observation, timestamp time.Time, originalAdapterKey string)
+	Get(params RequestParams) *Observation
+}
+
+// Config interface for configuration management
+type Config interface {
+	Set(string, any)
+	Get(string) any
+}
+
+// Server interface
+type Server interface {
+	Start() error
+	Close() error
+}

packages/streams-adapter/config/config.go

@@ -0,0 +1,64 @@
+package config
+
+import (
+	"os"
+	"strconv"
+)
+
+// Config holds all configuration for the adapter
+type Config struct {
+	HTTPPort      string
+	EAPort        string
+	EAHost        string
+	RedconPort    string
+	GoMetricsPort string
+
+	// Cache configuration
+	CacheMaxSize         uint // Maximum number of cache items (0 = default 10000)
+	CacheTTLMinutes      uint // Cache TTL in minutes (0 = default 5 minutes)
+	CacheCleanupInterval uint // Cache cleanup interval in minutes (0 = default 1 minute)
+
+	// Other configuration
+	LogLevel    string
+	AdapterName string
+}
+
+// Load reads configuration from environment variables
+func Load() *Config {
+	cfg := &Config{
+
+		HTTPPort:      getEnv("HTTP_PORT", "8080"),
+		EAPort:        getEnv("EA_PORT", "8070"),
+		EAHost:        getEnv("EA_INTERNAL_HOST", "localhost"),
+		RedconPort:    getEnv("REDCON_PORT", "6379"),
+		GoMetricsPort: getEnv("GO_METRICS_PORT", "9080"),
+
+		// Cache configuration
+		CacheMaxSize:         getEnvAsInt("CACHE_MAX_SIZE", 10000),
+		CacheTTLMinutes:      getEnvAsInt("CACHE_TTL_MINUTES", 5),
+		CacheCleanupInterval: getEnvAsInt("CACHE_CLEANUP_INTERVAL", 1),
+
+		// Other
+		LogLevel:    getEnv("LOG_LEVEL", "info"),
+		AdapterName: getEnv("ADAPTER_NAME", ""),
+	}
+
+	return cfg
+}
+
+// getEnv gets an environment variable with a default value
+func getEnv(key, defaultValue string) string {
+	if value := os.Getenv(key); value != "" {
+		return value
+	}
+	return defaultValue
+}
+
+// getEnvAsInt gets an environment variable as integer with a default value
+func getEnvAsInt(key string, defaultValue uint) uint {
+	valueStr := getEnv(key, "")
+	if value, err := strconv.ParseUint(valueStr, 10, 64); err == nil {
+		return uint(value)
+	}
+	return defaultValue
+}

packages/streams-adapter/go.mod

@@ -0,0 +1,5 @@
+module streams-adapter
+
+go 1.24.4
+
+require github.com/goccy/go-json v0.10.5

packages/streams-adapter/go.sum

@@ -0,0 +1,2 @@
+github.com/goccy/go-json v0.10.5 h1:Fq85nIqj+gXn/S5ahsiTlK3TmC85qgirsdTP/+DeaC4=
+github.com/goccy/go-json v0.10.5/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M=