feat(cache): 🎉 add message caching functionality

* Implement MessageCache to store events when no webhooks are available. * Add configuration options for enabling cache, setting data path, max age, and max events. * Create API endpoints for managing cached events, including listing, replaying, and deleting. * Integrate caching into the hooks manager to store events when no active webhooks are found. * Enhance logging for better traceability of cached events and operations.
2026-01-30 16:00:34 +02:00
parent 3901bbb668
commit c4d974d6ce
9 changed files with 1535 additions and 30 deletions
--- a/pkg/cache/message_cache.go
+++ b/pkg/cache/message_cache.go
@@ -0,0 +1,394 @@
+package cache
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sync"
+	"time"
+
+	"git.warky.dev/wdevs/whatshooked/pkg/events"
+	"git.warky.dev/wdevs/whatshooked/pkg/logging"
+)
+
+// CachedEvent represents an event stored in cache
+type CachedEvent struct {
+	ID          string       `json:"id"`
+	Event       events.Event `json:"event"`
+	Timestamp   time.Time    `json:"timestamp"`
+	Reason      string       `json:"reason"`
+	Attempts    int          `json:"attempts"`
+	LastAttempt *time.Time   `json:"last_attempt,omitempty"`
+}
+
+// MessageCache manages cached events when no webhooks are available
+type MessageCache struct {
+	events    map[string]*CachedEvent
+	mu        sync.RWMutex
+	dataPath  string
+	enabled   bool
+	maxAge    time.Duration // Maximum age before events are purged
+	maxEvents int           // Maximum number of events to keep
+}
+
+// Config holds cache configuration
+type Config struct {
+	Enabled   bool          `json:"enabled"`
+	DataPath  string        `json:"data_path"`
+	MaxAge    time.Duration `json:"max_age"`    // Default: 7 days
+	MaxEvents int           `json:"max_events"` // Default: 10000
+}
+
+// NewMessageCache creates a new message cache
+func NewMessageCache(cfg Config) (*MessageCache, error) {
+	if !cfg.Enabled {
+		return &MessageCache{
+			enabled: false,
+		}, nil
+	}
+
+	if cfg.DataPath == "" {
+		cfg.DataPath = "./data/cache"
+	}
+	if cfg.MaxAge == 0 {
+		cfg.MaxAge = 7 * 24 * time.Hour // 7 days
+	}
+	if cfg.MaxEvents == 0 {
+		cfg.MaxEvents = 10000
+	}
+
+	// Create cache directory
+	if err := os.MkdirAll(cfg.DataPath, 0755); err != nil {
+		return nil, fmt.Errorf("failed to create cache directory: %w", err)
+	}
+
+	cache := &MessageCache{
+		events:    make(map[string]*CachedEvent),
+		dataPath:  cfg.DataPath,
+		enabled:   true,
+		maxAge:    cfg.MaxAge,
+		maxEvents: cfg.MaxEvents,
+	}
+
+	// Load existing cached events
+	if err := cache.loadFromDisk(); err != nil {
+		logging.Warn("Failed to load cached events from disk", "error", err)
+	}
+
+	// Start cleanup goroutine
+	go cache.cleanupLoop()
+
+	logging.Info("Message cache initialized",
+		"enabled", cfg.Enabled,
+		"data_path", cfg.DataPath,
+		"max_age", cfg.MaxAge,
+		"max_events", cfg.MaxEvents)
+
+	return cache, nil
+}
+
+// Store adds an event to the cache
+func (c *MessageCache) Store(event events.Event, reason string) error {
+	if !c.enabled {
+		return nil
+	}
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	// Check if we're at capacity
+	if len(c.events) >= c.maxEvents {
+		// Remove oldest event
+		c.removeOldest()
+	}
+
+	// Generate unique ID
+	id := fmt.Sprintf("%d-%s", time.Now().UnixNano(), event.Type)
+
+	cached := &CachedEvent{
+		ID:        id,
+		Event:     event,
+		Timestamp: time.Now(),
+		Reason:    reason,
+		Attempts:  0,
+	}
+
+	c.events[id] = cached
+
+	// Save to disk asynchronously
+	go c.saveToDisk(cached)
+
+	logging.Debug("Event cached",
+		"event_id", id,
+		"event_type", event.Type,
+		"reason", reason,
+		"cache_size", len(c.events))
+
+	return nil
+}
+
+// Get retrieves a cached event by ID
+func (c *MessageCache) Get(id string) (*CachedEvent, bool) {
+	if !c.enabled {
+		return nil, false
+	}
+
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	event, exists := c.events[id]
+	return event, exists
+}
+
+// List returns all cached events
+func (c *MessageCache) List() []*CachedEvent {
+	if !c.enabled {
+		return nil
+	}
+
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	result := make([]*CachedEvent, 0, len(c.events))
+	for _, event := range c.events {
+		result = append(result, event)
+	}
+
+	return result
+}
+
+// ListByEventType returns cached events filtered by event type
+func (c *MessageCache) ListByEventType(eventType events.EventType) []*CachedEvent {
+	if !c.enabled {
+		return nil
+	}
+
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	result := make([]*CachedEvent, 0)
+	for _, cached := range c.events {
+		if cached.Event.Type == eventType {
+			result = append(result, cached)
+		}
+	}
+
+	return result
+}
+
+// Remove deletes an event from the cache
+func (c *MessageCache) Remove(id string) error {
+	if !c.enabled {
+		return nil
+	}
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if _, exists := c.events[id]; !exists {
+		return fmt.Errorf("cached event not found: %s", id)
+	}
+
+	delete(c.events, id)
+
+	// Remove from disk
+	go c.removeFromDisk(id)
+
+	logging.Debug("Event removed from cache", "event_id", id)
+
+	return nil
+}
+
+// IncrementAttempts increments the delivery attempt counter
+func (c *MessageCache) IncrementAttempts(id string) error {
+	if !c.enabled {
+		return nil
+	}
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	cached, exists := c.events[id]
+	if !exists {
+		return fmt.Errorf("cached event not found: %s", id)
+	}
+
+	now := time.Now()
+	cached.Attempts++
+	cached.LastAttempt = &now
+
+	// Update on disk
+	go c.saveToDisk(cached)
+
+	return nil
+}
+
+// Clear removes all cached events
+func (c *MessageCache) Clear() error {
+	if !c.enabled {
+		return nil
+	}
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	c.events = make(map[string]*CachedEvent)
+
+	// Clear disk cache
+	go c.clearDisk()
+
+	logging.Info("Message cache cleared")
+
+	return nil
+}
+
+// Count returns the number of cached events
+func (c *MessageCache) Count() int {
+	if !c.enabled {
+		return 0
+	}
+
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	return len(c.events)
+}
+
+// IsEnabled returns whether the cache is enabled
+func (c *MessageCache) IsEnabled() bool {
+	return c.enabled
+}
+
+// removeOldest removes the oldest event from the cache
+func (c *MessageCache) removeOldest() {
+	var oldestID string
+	var oldestTime time.Time
+
+	for id, cached := range c.events {
+		if oldestID == "" || cached.Timestamp.Before(oldestTime) {
+			oldestID = id
+			oldestTime = cached.Timestamp
+		}
+	}
+
+	if oldestID != "" {
+		delete(c.events, oldestID)
+		go c.removeFromDisk(oldestID)
+		logging.Debug("Removed oldest cached event due to capacity", "event_id", oldestID)
+	}
+}
+
+// cleanupLoop periodically removes expired events
+func (c *MessageCache) cleanupLoop() {
+	ticker := time.NewTicker(1 * time.Hour)
+	defer ticker.Stop()
+
+	for range ticker.C {
+		c.cleanup()
+	}
+}
+
+// cleanup removes expired events
+func (c *MessageCache) cleanup() {
+	if !c.enabled {
+		return
+	}
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	now := time.Now()
+	expiredIDs := make([]string, 0)
+
+	for id, cached := range c.events {
+		if now.Sub(cached.Timestamp) > c.maxAge {
+			expiredIDs = append(expiredIDs, id)
+		}
+	}
+
+	for _, id := range expiredIDs {
+		delete(c.events, id)
+		go c.removeFromDisk(id)
+	}
+
+	if len(expiredIDs) > 0 {
+		logging.Info("Cleaned up expired cached events", "count", len(expiredIDs))
+	}
+}
+
+// saveToDisk saves a cached event to disk
+func (c *MessageCache) saveToDisk(cached *CachedEvent) {
+	filePath := filepath.Join(c.dataPath, fmt.Sprintf("%s.json", cached.ID))
+
+	data, err := json.MarshalIndent(cached, "", "  ")
+	if err != nil {
+		logging.Error("Failed to marshal cached event", "event_id", cached.ID, "error", err)
+		return
+	}
+
+	if err := os.WriteFile(filePath, data, 0644); err != nil {
+		logging.Error("Failed to save cached event to disk", "event_id", cached.ID, "error", err)
+	}
+}
+
+// loadFromDisk loads all cached events from disk
+func (c *MessageCache) loadFromDisk() error {
+	files, err := filepath.Glob(filepath.Join(c.dataPath, "*.json"))
+	if err != nil {
+		return fmt.Errorf("failed to list cache files: %w", err)
+	}
+
+	loaded := 0
+	for _, file := range files {
+		data, err := os.ReadFile(file)
+		if err != nil {
+			logging.Warn("Failed to read cache file", "file", file, "error", err)
+			continue
+		}
+
+		var cached CachedEvent
+		if err := json.Unmarshal(data, &cached); err != nil {
+			logging.Warn("Failed to unmarshal cache file", "file", file, "error", err)
+			continue
+		}
+
+		// Skip expired events
+		if time.Since(cached.Timestamp) > c.maxAge {
+			os.Remove(file)
+			continue
+		}
+
+		c.events[cached.ID] = &cached
+		loaded++
+	}
+
+	if loaded > 0 {
+		logging.Info("Loaded cached events from disk", "count", loaded)
+	}
+
+	return nil
+}
+
+// removeFromDisk removes a cached event file from disk
+func (c *MessageCache) removeFromDisk(id string) {
+	filePath := filepath.Join(c.dataPath, fmt.Sprintf("%s.json", id))
+	if err := os.Remove(filePath); err != nil && !os.IsNotExist(err) {
+		logging.Error("Failed to remove cached event from disk", "event_id", id, "error", err)
+	}
+}
+
+// clearDisk removes all cache files from disk
+func (c *MessageCache) clearDisk() {
+	files, err := filepath.Glob(filepath.Join(c.dataPath, "*.json"))
+	if err != nil {
+		logging.Error("Failed to list cache files for clearing", "error", err)
+		return
+	}
+
+	for _, file := range files {
+		if err := os.Remove(file); err != nil {
+			logging.Error("Failed to remove cache file", "file", file, "error", err)
+		}
+	}
+}