From 4309b1918d7f0c0ab5bb9801398d071fa587d5ec Mon Sep 17 00:00:00 2001 From: Pierre Date: Fri, 13 Mar 2026 22:10:36 +0100 Subject: [PATCH] code cleanup --- README.md | 31 +-- docker/nginx.conf | 46 ++- docs/meta-core-architecture.md | 56 ++-- internal/api/server.go | 54 +--- internal/cache/handlers.go | 73 ----- internal/cache/index.go | 242 ---------------- internal/cache/invalidator.go | 166 ----------- internal/cache/manager.go | 494 --------------------------------- internal/cache/types.go | 48 ---- internal/config/config.go | 22 -- internal/webdav/handler.go | 45 +-- 11 files changed, 83 insertions(+), 1194 deletions(-) delete mode 100644 internal/cache/handlers.go delete mode 100644 internal/cache/index.go delete mode 100644 internal/cache/invalidator.go delete mode 100644 internal/cache/manager.go delete mode 100644 internal/cache/types.go diff --git a/README.md b/README.md index ae945d5..e3754ac 100644 --- a/README.md +++ b/README.md @@ -17,7 +17,7 @@ meta-core runs alongside each MetaMesh service (meta-sort, meta-fuse, meta-strem | **Metadata Storage** | Flat key-value schema with connection pooling and batch operations | | **File Watching** | Directory scanning, MidHash256 computation, event dispatch | | **Mount Management** | SMB/rclone configuration, health monitoring | -| **WebDAV Server** | File access with LRU caching and cache invalidation | +| **WebDAV Server** | File access with nginx proxy_cache | | **Event Publishing** | Redis keyspace notifications to meta:events stream | ### Design Characteristics @@ -112,9 +112,6 @@ docker run -v meta-core:/meta-core -v files:/files meta-core | `HEARTBEAT_INTERVAL_MS` | `30000` | Service heartbeat interval | | `STALE_THRESHOLD_MS` | `60000` | Stale service threshold | | `ENABLE_FILE_WATCHER` | `true` | Enable file scanning | -| `CACHE_ENABLED` | `true` | Enable WebDAV caching | -| `CACHE_MAX_SIZE_GB` | `100` | Maximum cache size in GB | -| `CACHE_TTL_SECONDS` | `3600` | Cache entry TTL | ## API Reference @@ -224,20 +221,6 @@ curl http://localhost:9000/api/events/stream # data: {"type":"add","path":"movies/Movie.mkv","midhash256":"bafkr..."} ``` -### Cache Management - -```bash -# Get cache status -curl http://localhost:9000/api/cache/status -# {"enabled":true,"size":"2.5GB","entries":150,"maxSize":"100GB"} - -# Clear cache -curl -X POST http://localhost:9000/api/cache/clear - -# Detailed cache stats -curl http://localhost:9000/api/cache/stats -``` - ### KV Browser ```bash @@ -465,7 +448,6 @@ make clean | `internal/storage` | Redis client wrapper with flat key-value operations | | `internal/discovery` | Service registration, heartbeat loop, discovery | | `internal/api` | HTTP server, router, and all endpoint handlers | -| `internal/cache` | WebDAV file caching with LRU eviction and invalidation | | `internal/events` | Redis keyspace notification publisher to meta:events | | `internal/watcher` | File system scanning, MidHash256 computation, event dispatch | | `internal/watchers` | Polling-based watcher configuration management | @@ -482,12 +464,11 @@ make clean 5. Start election (acquire lock or become follower) 6. Start service discovery (register + heartbeat loop) 7. Initialize dead service cleaner -8. Initialize cache manager (if enabled) -9. Initialize file watcher (if enabled) -10. Initialize mount manager -11. Start HTTP API server -12. Wait for SIGINT/SIGTERM -13. Shutdown in reverse order +8. Initialize file watcher (if enabled) +9. Initialize mount manager +10. Start HTTP API server (WebDAV caching handled by nginx) +11. Wait for SIGINT/SIGTERM +12. Shutdown in reverse order ``` ## Integration diff --git a/docker/nginx.conf b/docker/nginx.conf index 5dd1dc3..66ae59f 100644 --- a/docker/nginx.conf +++ b/docker/nginx.conf @@ -22,6 +22,14 @@ http { gzip on; gzip_types text/plain text/css application/json application/javascript text/xml application/xml; + # WebDAV cache zone definition + proxy_cache_path /meta-core/cache + levels=1:2 # 2-level directory structure + keys_zone=webdav_cache:10m # 10MB shared memory for keys + max_size=10g # Max 10GB on disk + inactive=1h # Remove if not accessed for 1 hour + use_temp_path=off; # Write directly to cache dir + # Upstream: meta-core Go API upstream meta-core-api { server 127.0.0.1:9000; @@ -192,11 +200,33 @@ http { } # ========================================================================= - # WebDAV endpoint for file access (proxied to Go for caching) + # WebDAV endpoint for file access (nginx proxy_cache for GET/HEAD) # Supports: GET (read), PUT (write), DELETE, MKCOL (mkdir), PROPFIND - # Files are cached by the Go service for faster repeated access + # Files are cached by nginx for faster repeated access # ========================================================================= location /webdav/ { + # Slice module for large file chunked caching + slice 1m; + proxy_cache_key $uri$slice_range; + proxy_set_header Range $slice_range; + + # Enable caching for GET/HEAD requests + proxy_cache webdav_cache; + proxy_cache_valid 200 206 1h; + proxy_cache_methods GET HEAD; + + # Don't cache write operations + set $no_cache 0; + if ($request_method !~ ^(GET|HEAD)$) { + set $no_cache 1; + } + proxy_no_cache $no_cache; + proxy_cache_bypass $no_cache; + + # Cache status header for debugging + add_header X-Cache-Status $upstream_cache_status always; + + # Proxy settings proxy_pass http://meta-core-api/webdav/; proxy_http_version 1.1; proxy_set_header Host $host; @@ -204,14 +234,12 @@ http { proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; - # Disable buffering for streaming and large files - proxy_buffering off; - proxy_request_buffering off; + # Enable buffering for caching (required for proxy_cache) + proxy_buffering on; + proxy_request_buffering off; # Don't buffer uploads - # No size limit for uploads (large media files) + # Large file support client_max_body_size 0; - - # Timeout settings for large files proxy_read_timeout 600s; proxy_send_timeout 600s; @@ -219,7 +247,7 @@ http { add_header 'Access-Control-Allow-Origin' '*' always; add_header 'Access-Control-Allow-Methods' 'GET, HEAD, PUT, DELETE, MKCOL, COPY, MOVE, PROPFIND, OPTIONS' always; add_header 'Access-Control-Allow-Headers' 'Authorization, Content-Type, Depth, Destination, Range, If-Range' always; - add_header 'Access-Control-Expose-Headers' 'Content-Length, Content-Range, Accept-Ranges' always; + add_header 'Access-Control-Expose-Headers' 'Content-Length, Content-Range, Accept-Ranges, X-Cache-Status' always; # Handle OPTIONS preflight if ($request_method = 'OPTIONS') { diff --git a/docs/meta-core-architecture.md b/docs/meta-core-architecture.md index 99712f5..91e29e3 100644 --- a/docs/meta-core-architecture.md +++ b/docs/meta-core-architecture.md @@ -262,34 +262,37 @@ Follower meta-core instances: ### WebDAV Caching Architecture -Meta-core provides a WebDAV server at `/webdav/` with integrated caching: +Meta-core provides a WebDAV server at `/webdav/` with nginx proxy_cache for file caching: ``` ┌─────────────────────────────────────────────────────────────────┐ -│ WebDAV Server │ +│ nginx (:80) │ │ │ │ ┌──────────────┐ ┌──────────────┐ ┌────────────────────────┐ │ -│ │ HTTP Handler │─►│ Cache Layer │─►│ File System (/files) │ │ -│ │ │ │ │ │ │ │ -│ │ GET/PUT/ │ │ LRU Index │ │ - watch/ │ │ -│ │ DELETE/MKCOL │ │ TTL Expiry │ │ - plugin/ │ │ -│ │ │ │ Invalidation │ │ - corn/ │ │ -│ └──────────────┘ └──────────────┘ └────────────────────────┘ │ -│ │ │ -│ ▼ │ -│ ┌──────────────┐ │ -│ │ Redis │ │ -│ │ Keyspace │ │ -│ │ Notifications│ │ -│ └──────────────┘ │ +│ │ HTTP Request │─►│ proxy_cache │─►│ Go WebDAV Handler │ │ +│ │ │ │ │ │ (:9000) │ │ +│ │ GET/HEAD │ │ Cached ─────►│ │ │ │ +│ │ PUT/DELETE │ │ Bypass ─────►│ │ │ │ +│ │ MKCOL │ │ │ │ │ │ +│ └──────────────┘ └──────────────┘ └──────────────────────┐ │ +│ │ │ │ +│ /meta-core/cache/ │ ▼ │ +│ (nginx managed) │ ┌────────────────────┐│ +│ │ │ File System ││ +│ │ │ (/files) ││ +│ │ │ - watch/ ││ +│ │ │ - plugin/ ││ +│ │ │ - corn/ ││ +│ │ └────────────────────┘│ └─────────────────────────────────────────────────────────────────┘ ``` Cache features: -- **LRU Index**: Tracks file access times and sizes for eviction -- **Eviction**: Removes least-recently-used entries when limit exceeded -- **TTL**: Expires entries after configured time (`CACHE_TTL_SECONDS`) -- **Invalidation**: Cache invalidator listens to Redis keyspace changes +- **proxy_cache**: nginx handles caching of GET/HEAD requests with 1-hour TTL +- **Slice module**: Large files cached in 1MB chunks for efficient range requests +- **TTL-based expiry**: Entries expire after 1 hour of inactivity +- **X-Cache-Status header**: Returns MISS/HIT/EXPIRED for debugging +- **Write bypass**: PUT/DELETE/MKCOL operations bypass cache - **Directory Listing**: Returns JSON for programmatic access ### File Watching System @@ -444,11 +447,6 @@ File Watching POST /api/scan/trigger → Trigger rescan GET /api/events/stream → Stream file events (SSE) -Cache Management - GET /api/cache/status → Cache statistics - POST /api/cache/clear → Clear cache - GET /api/cache/stats → Detailed metrics - KV Browser GET /api/kv/info → Storage statistics GET /api/kv/keys → Key listing with cursor @@ -643,12 +641,7 @@ meta-core │ │ ├── data.go # Data endpoints │ │ ├── services.go # Service discovery endpoints │ │ ├── mounts.go # Mount management endpoints -│ │ ├── cache.go # Cache management endpoints │ │ └── kv.go # KV browser endpoints -│ ├── cache/ # WebDAV caching with LRU -│ │ ├── lru.go # LRU index implementation -│ │ ├── invalidator.go # Keyspace notification listener -│ │ └── manager.go # Cache lifecycle management │ ├── config/ # Configuration loading │ │ └── config.go │ ├── discovery/ # Service registration @@ -1034,9 +1027,6 @@ Meta-core exposes two WebDAV URLs via the `/urls` API to support both internal a | `META_CORE_CACHE_ENABLED` | `false` | Enable local read cache | | `META_CORE_CACHE_TTL` | `60` | Cache TTL (seconds) | | `ENABLE_FILE_WATCHER` | `true` | Enable file scanning | -| `CACHE_ENABLED` | `true` | Enable WebDAV caching | -| `CACHE_MAX_SIZE_GB` | `100` | Max cache size | -| `CACHE_TTL_SECONDS` | `3600` | Cache entry TTL | ### File Paths @@ -1048,7 +1038,7 @@ Meta-core exposes two WebDAV URLs via the `/urls` API to support both internal a | `/meta-core/db/redis/appendonly.aof` | Redis AOF log | | `/meta-core/services/*.json` | Service registry files | | `/meta-core/mounts/mounts.json` | Mount configurations | -| `/meta-core/cache/` | Cached files directory | +| `/meta-core/cache/` | nginx proxy_cache directory | | `/meta-core/watchers.json` | Watcher configurations | --- diff --git a/internal/api/server.go b/internal/api/server.go index a0aba66..ff31694 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -8,7 +8,6 @@ import ( "time" "github.com/gorilla/mux" - "github.com/metazla/meta-core/internal/cache" "github.com/metazla/meta-core/internal/config" "github.com/metazla/meta-core/internal/discovery" "github.com/metazla/meta-core/internal/events" @@ -36,9 +35,6 @@ type Server struct { watchersManager *watchers.Manager watchersPoller *watchers.Poller watchersHandlers *watchers.Handlers - cacheManager *cache.Manager - cacheHandlers *cache.Handlers - cacheInvalidator *cache.Invalidator metaPublisher *events.MetaPublisher webdavHandler *webdav.Handler router *mux.Router @@ -109,17 +105,8 @@ func NewServer( mountsManager.SetPoller(s.mountPoller) } - // Initialize cache manager - cacheManager, err := cache.NewManager(cfg) - if err != nil { - log.Printf("[API] Warning: failed to initialize cache manager: %v", err) - } else { - s.cacheManager = cacheManager - s.cacheHandlers = cache.NewHandlers(cacheManager) - s.webdavHandler = webdav.NewHandler(cfg, cacheManager) - - // Cache invalidator will be initialized in Start() after Redis is connected - } + // Initialize WebDAV handler (caching is handled by nginx proxy_cache) + s.webdavHandler = webdav.NewHandler(cfg) s.setupRoutes() return s @@ -201,13 +188,7 @@ func (s *Server) setupRoutes() { log.Println("[API] Watchers management routes registered at /api/watchers/*") } - // Cache management routes (if cache initialized) - if s.cacheHandlers != nil { - s.cacheHandlers.RegisterRoutes(s.router) - log.Println("[API] Cache management routes registered at /api/cache/*") - } - - // WebDAV handler (if cache initialized) + // WebDAV handler (caching is handled by nginx proxy_cache) if s.webdavHandler != nil { s.router.PathPrefix("/webdav/").Handler(s.webdavHandler) s.router.PathPrefix("/webdav").Handler(s.webdavHandler) @@ -253,21 +234,6 @@ func (s *Server) Start() error { } } - // Start cache manager (if initialized) - if s.cacheManager != nil { - if err := s.cacheManager.Start(); err != nil { - log.Printf("[API] Warning: failed to start cache manager: %v", err) - } - - // Initialize cache invalidator now that Redis is connected - if s.storage != nil && s.storage.IsConnected() { - s.cacheInvalidator = cache.NewInvalidator(s.cacheManager, s.storage.GetRedisClient()) - if err := s.cacheInvalidator.Start(); err != nil { - log.Printf("[API] Warning: failed to start cache invalidator: %v", err) - } - } - } - // Start meta publisher (if storage is connected) // Publishes keyspace notifications to meta:events stream if s.storage != nil && s.storage.IsConnected() { @@ -313,20 +279,6 @@ func (s *Server) Stop() error { } } - // Stop cache invalidator - if s.cacheInvalidator != nil { - if err := s.cacheInvalidator.Stop(); err != nil { - log.Printf("[API] Warning: failed to stop cache invalidator: %v", err) - } - } - - // Stop cache manager - if s.cacheManager != nil { - if err := s.cacheManager.Stop(); err != nil { - log.Printf("[API] Warning: failed to stop cache manager: %v", err) - } - } - if s.server == nil { return nil } diff --git a/internal/cache/handlers.go b/internal/cache/handlers.go deleted file mode 100644 index af8404a..0000000 --- a/internal/cache/handlers.go +++ /dev/null @@ -1,73 +0,0 @@ -package cache - -import ( - "encoding/json" - "log" - "net/http" - - "github.com/gorilla/mux" -) - -// Handlers provides HTTP handlers for cache management -type Handlers struct { - manager *Manager -} - -// NewHandlers creates new cache handlers -func NewHandlers(manager *Manager) *Handlers { - return &Handlers{ - manager: manager, - } -} - -// RegisterRoutes registers cache management routes -func (h *Handlers) RegisterRoutes(router *mux.Router) { - router.HandleFunc("/api/cache/stats", h.handleStats).Methods("GET") - router.HandleFunc("/api/cache/clear", h.handleClear).Methods("POST") - router.HandleFunc("/api/cache/invalidate", h.handleInvalidate).Methods("DELETE") -} - -// handleStats returns cache statistics -func (h *Handlers) handleStats(w http.ResponseWriter, r *http.Request) { - stats := h.manager.Stats() - - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(stats) -} - -// handleClear clears the entire cache -func (h *Handlers) handleClear(w http.ResponseWriter, r *http.Request) { - if err := h.manager.Clear(); err != nil { - log.Printf("[CacheHandlers] Error clearing cache: %v", err) - http.Error(w, err.Error(), http.StatusInternalServerError) - return - } - - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(map[string]interface{}{ - "success": true, - "message": "Cache cleared", - }) -} - -// handleInvalidate invalidates a specific path -func (h *Handlers) handleInvalidate(w http.ResponseWriter, r *http.Request) { - path := r.URL.Query().Get("path") - if path == "" { - http.Error(w, "path query parameter required", http.StatusBadRequest) - return - } - - if err := h.manager.Invalidate(path); err != nil { - log.Printf("[CacheHandlers] Error invalidating path %s: %v", path, err) - http.Error(w, err.Error(), http.StatusInternalServerError) - return - } - - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(map[string]interface{}{ - "success": true, - "message": "Path invalidated", - "path": path, - }) -} diff --git a/internal/cache/index.go b/internal/cache/index.go deleted file mode 100644 index 1970d7e..0000000 --- a/internal/cache/index.go +++ /dev/null @@ -1,242 +0,0 @@ -package cache - -import ( - "container/list" - "sync" - "time" -) - -// LRUIndex maintains an in-memory LRU index of cache entries -type LRUIndex struct { - mu sync.RWMutex - entries map[string]*list.Element // path -> list element - lruList *list.List // LRU order (front = most recent) - maxSize int64 // Maximum total size in bytes - currSize int64 // Current total size in bytes - - // Stats - hitCount int64 - missCount int64 - evictionCount int64 -} - -// lruEntry wraps CacheEntry for the linked list -type lruEntry struct { - entry *CacheEntry -} - -// NewLRUIndex creates a new LRU index -func NewLRUIndex(maxSizeGB int) *LRUIndex { - return &LRUIndex{ - entries: make(map[string]*list.Element), - lruList: list.New(), - maxSize: int64(maxSizeGB) * 1024 * 1024 * 1024, // GB to bytes - } -} - -// Get retrieves a cache entry and updates its LRU position -// Returns nil if not found -func (idx *LRUIndex) Get(path string) *CacheEntry { - idx.mu.Lock() - defer idx.mu.Unlock() - - elem, exists := idx.entries[path] - if !exists { - idx.missCount++ - return nil - } - - // Move to front (most recently used) - idx.lruList.MoveToFront(elem) - - // Update last accessed time - entry := elem.Value.(*lruEntry).entry - entry.LastAccessed = time.Now() - - idx.hitCount++ - return entry -} - -// Put adds or updates a cache entry -func (idx *LRUIndex) Put(entry *CacheEntry) { - idx.mu.Lock() - defer idx.mu.Unlock() - - if elem, exists := idx.entries[entry.OriginalPath]; exists { - // Update existing entry - oldEntry := elem.Value.(*lruEntry).entry - idx.currSize -= oldEntry.Size - idx.currSize += entry.Size - - elem.Value.(*lruEntry).entry = entry - idx.lruList.MoveToFront(elem) - } else { - // Add new entry - le := &lruEntry{entry: entry} - elem := idx.lruList.PushFront(le) - idx.entries[entry.OriginalPath] = elem - idx.currSize += entry.Size - } -} - -// Remove removes a cache entry by path -// Returns the removed entry or nil if not found -func (idx *LRUIndex) Remove(path string) *CacheEntry { - idx.mu.Lock() - defer idx.mu.Unlock() - - elem, exists := idx.entries[path] - if !exists { - return nil - } - - entry := elem.Value.(*lruEntry).entry - idx.lruList.Remove(elem) - delete(idx.entries, path) - idx.currSize -= entry.Size - - return entry -} - -// GetLRUEntry returns the least recently used entry without removing it -// Returns nil if the index is empty -func (idx *LRUIndex) GetLRUEntry() *CacheEntry { - idx.mu.RLock() - defer idx.mu.RUnlock() - - back := idx.lruList.Back() - if back == nil { - return nil - } - - return back.Value.(*lruEntry).entry -} - -// PopLRUEntry removes and returns the least recently used entry -// Returns nil if the index is empty -func (idx *LRUIndex) PopLRUEntry() *CacheEntry { - idx.mu.Lock() - defer idx.mu.Unlock() - - back := idx.lruList.Back() - if back == nil { - return nil - } - - entry := back.Value.(*lruEntry).entry - idx.lruList.Remove(back) - delete(idx.entries, entry.OriginalPath) - idx.currSize -= entry.Size - idx.evictionCount++ - - return entry -} - -// GetExpiredEntries returns all entries that have exceeded the TTL -func (idx *LRUIndex) GetExpiredEntries(ttl time.Duration) []*CacheEntry { - idx.mu.RLock() - defer idx.mu.RUnlock() - - var expired []*CacheEntry - cutoff := time.Now().Add(-ttl) - - for elem := idx.lruList.Back(); elem != nil; elem = elem.Prev() { - entry := elem.Value.(*lruEntry).entry - if entry.CachedAt.Before(cutoff) { - expired = append(expired, entry) - } - } - - return expired -} - -// Size returns the current total size of cached files -func (idx *LRUIndex) Size() int64 { - idx.mu.RLock() - defer idx.mu.RUnlock() - return idx.currSize -} - -// Count returns the number of entries in the index -func (idx *LRUIndex) Count() int { - idx.mu.RLock() - defer idx.mu.RUnlock() - return len(idx.entries) -} - -// MaxSize returns the maximum cache size -func (idx *LRUIndex) MaxSize() int64 { - return idx.maxSize -} - -// NeedsEviction returns true if the cache size exceeds the maximum -func (idx *LRUIndex) NeedsEviction() bool { - idx.mu.RLock() - defer idx.mu.RUnlock() - return idx.currSize > idx.maxSize -} - -// Stats returns cache statistics -func (idx *LRUIndex) Stats() *CacheStats { - idx.mu.RLock() - defer idx.mu.RUnlock() - - total := idx.hitCount + idx.missCount - var hitRate float64 - if total > 0 { - hitRate = float64(idx.hitCount) / float64(total) * 100 - } - - return &CacheStats{ - Enabled: true, - TotalSize: idx.currSize, - MaxSize: idx.maxSize, - EntryCount: len(idx.entries), - HitCount: idx.hitCount, - MissCount: idx.missCount, - EvictionCount: idx.evictionCount, - HitRate: hitRate, - } -} - -// GetAllEntries returns all cache entries (for persistence) -func (idx *LRUIndex) GetAllEntries() map[string]*CacheEntry { - idx.mu.RLock() - defer idx.mu.RUnlock() - - result := make(map[string]*CacheEntry, len(idx.entries)) - for path, elem := range idx.entries { - result[path] = elem.Value.(*lruEntry).entry - } - return result -} - -// LoadFromMap loads entries from a map (for loading persisted index) -func (idx *LRUIndex) LoadFromMap(entries map[string]*CacheEntry) { - idx.mu.Lock() - defer idx.mu.Unlock() - - // Clear existing entries - idx.lruList.Init() - idx.entries = make(map[string]*list.Element) - idx.currSize = 0 - - // Load entries sorted by LastAccessed (oldest first for proper LRU order) - // For simplicity, we just add them in arbitrary order - the access pattern will sort them - for _, entry := range entries { - le := &lruEntry{entry: entry} - elem := idx.lruList.PushBack(le) // Add to back (oldest) - idx.entries[entry.OriginalPath] = elem - idx.currSize += entry.Size - } -} - -// Clear removes all entries from the index -func (idx *LRUIndex) Clear() { - idx.mu.Lock() - defer idx.mu.Unlock() - - idx.lruList.Init() - idx.entries = make(map[string]*list.Element) - idx.currSize = 0 -} diff --git a/internal/cache/invalidator.go b/internal/cache/invalidator.go deleted file mode 100644 index edcad6c..0000000 --- a/internal/cache/invalidator.go +++ /dev/null @@ -1,166 +0,0 @@ -package cache - -import ( - "context" - "log" - "sync" - "time" - - "github.com/redis/go-redis/v9" -) - -const ( - // EventsStream is the Redis Stream name for file events - EventsStream = "file:events" - - // ConsumerGroup is the consumer group name for cache invalidation - ConsumerGroup = "cache-invalidator" - - // ConsumerName identifies this consumer in the group - ConsumerName = "meta-core-cache" -) - -// Invalidator subscribes to Redis Stream for file events and invalidates cache -type Invalidator struct { - manager *Manager - redisClient *redis.Client - ctx context.Context - cancel context.CancelFunc - wg sync.WaitGroup - lastID string -} - -// NewInvalidator creates a new cache invalidator -func NewInvalidator(manager *Manager, redisClient *redis.Client) *Invalidator { - ctx, cancel := context.WithCancel(context.Background()) - return &Invalidator{ - manager: manager, - redisClient: redisClient, - ctx: ctx, - cancel: cancel, - lastID: "$", // Start from new messages - } -} - -// Start starts listening to the Redis Stream -func (inv *Invalidator) Start() error { - if inv.manager == nil || !inv.manager.Enabled() { - log.Println("[CacheInvalidator] Cache disabled, not starting invalidator") - return nil - } - - if inv.redisClient == nil { - log.Println("[CacheInvalidator] Redis client not available, not starting invalidator") - return nil - } - - // Try to create consumer group (ignore error if already exists) - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - - err := inv.redisClient.XGroupCreateMkStream(ctx, EventsStream, ConsumerGroup, "$").Err() - if err != nil && err.Error() != "BUSYGROUP Consumer Group name already exists" { - // Log but don't fail - we can still use XREAD - log.Printf("[CacheInvalidator] Note: consumer group creation: %v (using XREAD fallback)", err) - } - - inv.wg.Add(1) - go inv.listen() - - log.Println("[CacheInvalidator] Started listening on file:events stream") - return nil -} - -// Stop stops the invalidator -func (inv *Invalidator) Stop() error { - log.Println("[CacheInvalidator] Stopping...") - inv.cancel() - inv.wg.Wait() - return nil -} - -// listen reads from the Redis Stream and processes events -func (inv *Invalidator) listen() { - defer inv.wg.Done() - - for { - select { - case <-inv.ctx.Done(): - return - default: - inv.readEvents() - } - } -} - -// readEvents reads and processes a batch of events -func (inv *Invalidator) readEvents() { - ctx, cancel := context.WithTimeout(inv.ctx, 5*time.Second) - defer cancel() - - // Use XREAD with BLOCK for efficient waiting - streams, err := inv.redisClient.XRead(ctx, &redis.XReadArgs{ - Streams: []string{EventsStream, inv.lastID}, - Count: 100, - Block: 2 * time.Second, - }).Result() - - if err != nil { - if err == context.Canceled || err == context.DeadlineExceeded || err == redis.Nil { - return // Normal timeout or shutdown - } - log.Printf("[CacheInvalidator] Error reading stream: %v", err) - time.Sleep(1 * time.Second) // Backoff on error - return - } - - for _, stream := range streams { - for _, msg := range stream.Messages { - inv.processEvent(msg) - inv.lastID = msg.ID - } - } -} - -// processEvent processes a single file event -func (inv *Invalidator) processEvent(msg redis.XMessage) { - eventType, _ := msg.Values["type"].(string) - path, _ := msg.Values["path"].(string) - - switch eventType { - case "delete": - // File deleted - invalidate cache - if path != "" { - log.Printf("[CacheInvalidator] Invalidating deleted file: %s", path) - inv.manager.Invalidate(path) - } - - case "change": - // File changed - invalidate cache - if path != "" { - log.Printf("[CacheInvalidator] Invalidating changed file: %s", path) - inv.manager.Invalidate(path) - } - - case "rename": - // File renamed - invalidate old path - oldPath, _ := msg.Values["oldPath"].(string) - if oldPath != "" { - log.Printf("[CacheInvalidator] Invalidating renamed file (old): %s", oldPath) - inv.manager.Invalidate(oldPath) - } - // Also invalidate new path in case it was cached before - if path != "" { - inv.manager.Invalidate(path) - } - - case "reset": - // Full reset - clear entire cache - log.Println("[CacheInvalidator] Reset event received - clearing cache") - inv.manager.Clear() - - case "add": - // New file - nothing to invalidate - // (file wouldn't be in cache yet) - } -} diff --git a/internal/cache/manager.go b/internal/cache/manager.go deleted file mode 100644 index 85c027d..0000000 --- a/internal/cache/manager.go +++ /dev/null @@ -1,494 +0,0 @@ -package cache - -import ( - "context" - "crypto/sha256" - "encoding/hex" - "encoding/json" - "fmt" - "io" - "log" - "net/http" - "os" - "path/filepath" - "strings" - "sync" - "time" - - "github.com/metazla/meta-core/internal/config" -) - -// Manager handles caching operations for WebDAV files -type Manager struct { - config *config.Config - index *LRUIndex - filesPath string - cachePath string - ttl time.Duration - enabled bool - - mu sync.RWMutex - ctx context.Context - cancel context.CancelFunc - cleanupWg sync.WaitGroup -} - -// NewManager creates a new cache manager -func NewManager(cfg *config.Config) (*Manager, error) { - if !cfg.CacheEnabled { - log.Println("[Cache] Caching is disabled") - return &Manager{ - config: cfg, - enabled: false, - }, nil - } - - // Ensure cache directory exists - if err := os.MkdirAll(cfg.CacheDir(), 0755); err != nil { - return nil, fmt.Errorf("failed to create cache directory: %w", err) - } - - ctx, cancel := context.WithCancel(context.Background()) - - m := &Manager{ - config: cfg, - index: NewLRUIndex(cfg.CacheMaxSizeGB), - filesPath: cfg.FilesPath, - cachePath: cfg.CacheDir(), - ttl: time.Duration(cfg.CacheTTLSeconds) * time.Second, - enabled: true, - ctx: ctx, - cancel: cancel, - } - - // Load existing index from disk - if err := m.loadIndex(); err != nil { - log.Printf("[Cache] Warning: failed to load index: %v (starting fresh)", err) - } - - return m, nil -} - -// Start starts the cache manager (cleanup goroutine) -func (m *Manager) Start() error { - if !m.enabled { - return nil - } - - log.Printf("[Cache] Starting cache manager (max: %dGB, ttl: %v)", - m.config.CacheMaxSizeGB, m.ttl) - - // Start cleanup goroutine - m.cleanupWg.Add(1) - go m.cleanupLoop() - - return nil -} - -// Stop stops the cache manager -func (m *Manager) Stop() error { - if !m.enabled { - return nil - } - - log.Println("[Cache] Stopping cache manager...") - m.cancel() - m.cleanupWg.Wait() - - // Persist index - if err := m.saveIndex(); err != nil { - log.Printf("[Cache] Warning: failed to save index: %v", err) - } - - return nil -} - -// Enabled returns whether caching is enabled -func (m *Manager) Enabled() bool { - return m.enabled -} - -// Get retrieves a file from cache or fetches it from origin -// Returns the cached file path and content type, or an error -func (m *Manager) Get(path string) (string, string, error) { - if !m.enabled { - // Return original path if caching disabled - return filepath.Join(m.filesPath, path), "", nil - } - - // Normalize path (remove leading slash if present) - path = strings.TrimPrefix(path, "/") - - // Check cache - entry := m.index.Get(path) - if entry != nil { - // Verify cached file still exists - if _, err := os.Stat(entry.CachePath); err == nil { - log.Printf("[Cache] HIT: %s", path) - return entry.CachePath, entry.ContentType, nil - } - // Cached file missing, remove from index - m.index.Remove(path) - } - - log.Printf("[Cache] MISS: %s", path) - - // Fetch from origin and cache - cachedPath, contentType, err := m.fetchAndCache(path) - if err != nil { - return "", "", err - } - - return cachedPath, contentType, nil -} - -// GetOriginalPath returns the original file path (for when caching is bypassed) -func (m *Manager) GetOriginalPath(path string) string { - path = strings.TrimPrefix(path, "/") - return filepath.Join(m.filesPath, path) -} - -// fetchAndCache fetches a file from origin and caches it -func (m *Manager) fetchAndCache(path string) (string, string, error) { - originPath := filepath.Join(m.filesPath, path) - - // Check if origin file exists - info, err := os.Stat(originPath) - if err != nil { - return "", "", err - } - - // Skip caching directories - if info.IsDir() { - return originPath, "", nil - } - - // Generate cache path using SHA256 of original path with 2-char sharding - cachePath := m.getCachePath(path) - - // Ensure cache subdirectory exists - cacheDir := filepath.Dir(cachePath) - if err := os.MkdirAll(cacheDir, 0755); err != nil { - return "", "", fmt.Errorf("failed to create cache dir: %w", err) - } - - // Copy file to cache - src, err := os.Open(originPath) - if err != nil { - return "", "", err - } - defer src.Close() - - dst, err := os.Create(cachePath) - if err != nil { - return "", "", err - } - defer dst.Close() - - _, err = io.Copy(dst, src) - if err != nil { - os.Remove(cachePath) - return "", "", err - } - - // Detect content type - contentType := detectContentType(originPath) - - // Create cache entry - entry := &CacheEntry{ - OriginalPath: path, - CachePath: cachePath, - Size: info.Size(), - CachedAt: time.Now(), - LastAccessed: time.Now(), - ContentType: contentType, - } - - // Add to index - m.index.Put(entry) - - // Trigger eviction if needed - m.evictIfNeeded() - - log.Printf("[Cache] Cached: %s (%d bytes)", path, info.Size()) - - return cachePath, contentType, nil -} - -// getCachePath generates a cache path using SHA256 with 2-char directory sharding -func (m *Manager) getCachePath(path string) string { - hash := sha256.Sum256([]byte(path)) - hexHash := hex.EncodeToString(hash[:]) - - // Use first 2 characters for sharding - shard := hexHash[:2] - filename := hexHash[2:] - - return filepath.Join(m.cachePath, shard, filename) -} - -// Invalidate removes a specific path from the cache -func (m *Manager) Invalidate(path string) error { - if !m.enabled { - return nil - } - - path = strings.TrimPrefix(path, "/") - - entry := m.index.Remove(path) - if entry == nil { - return nil // Not in cache - } - - // Remove cached file - if err := os.Remove(entry.CachePath); err != nil && !os.IsNotExist(err) { - log.Printf("[Cache] Warning: failed to remove cached file: %v", err) - } - - log.Printf("[Cache] Invalidated: %s", path) - return nil -} - -// Clear removes all entries from the cache -func (m *Manager) Clear() error { - if !m.enabled { - return nil - } - - m.mu.Lock() - defer m.mu.Unlock() - - // Get all entries before clearing - entries := m.index.GetAllEntries() - - // Clear index - m.index.Clear() - - // Remove all cached files - for _, entry := range entries { - if err := os.Remove(entry.CachePath); err != nil && !os.IsNotExist(err) { - log.Printf("[Cache] Warning: failed to remove cached file: %v", err) - } - } - - // Remove empty shard directories - files, err := os.ReadDir(m.cachePath) - if err == nil { - for _, f := range files { - if f.IsDir() && len(f.Name()) == 2 { - dirPath := filepath.Join(m.cachePath, f.Name()) - os.Remove(dirPath) // Only removes if empty - } - } - } - - log.Printf("[Cache] Cleared %d entries", len(entries)) - return nil -} - -// Stats returns cache statistics -func (m *Manager) Stats() *CacheStats { - if !m.enabled { - return &CacheStats{Enabled: false} - } - - stats := m.index.Stats() - stats.TTLSeconds = m.config.CacheTTLSeconds - return stats -} - -// evictIfNeeded evicts LRU entries until cache is under max size -func (m *Manager) evictIfNeeded() { - for m.index.NeedsEviction() { - entry := m.index.PopLRUEntry() - if entry == nil { - break - } - - // Remove cached file - if err := os.Remove(entry.CachePath); err != nil && !os.IsNotExist(err) { - log.Printf("[Cache] Warning: failed to remove evicted file: %v", err) - } - - log.Printf("[Cache] Evicted (LRU): %s", entry.OriginalPath) - } -} - -// cleanupLoop periodically cleans up expired entries -func (m *Manager) cleanupLoop() { - defer m.cleanupWg.Done() - - ticker := time.NewTicker(5 * time.Minute) - defer ticker.Stop() - - for { - select { - case <-m.ctx.Done(): - return - case <-ticker.C: - m.cleanupExpired() - m.evictIfNeeded() - - // Periodically save index - if err := m.saveIndex(); err != nil { - log.Printf("[Cache] Warning: failed to save index: %v", err) - } - } - } -} - -// cleanupExpired removes entries that have exceeded TTL -func (m *Manager) cleanupExpired() { - expired := m.index.GetExpiredEntries(m.ttl) - - for _, entry := range expired { - m.index.Remove(entry.OriginalPath) - - if err := os.Remove(entry.CachePath); err != nil && !os.IsNotExist(err) { - log.Printf("[Cache] Warning: failed to remove expired file: %v", err) - } - - log.Printf("[Cache] Expired (TTL): %s", entry.OriginalPath) - } - - if len(expired) > 0 { - log.Printf("[Cache] Cleaned up %d expired entries", len(expired)) - } -} - -// loadIndex loads the cache index from disk -func (m *Manager) loadIndex() error { - indexPath := m.config.CacheIndexPath() - - data, err := os.ReadFile(indexPath) - if err != nil { - if os.IsNotExist(err) { - return nil // No index file yet - } - return err - } - - var idx CacheIndex - if err := json.Unmarshal(data, &idx); err != nil { - return err - } - - // Validate and load entries - validEntries := make(map[string]*CacheEntry) - for path, entry := range idx.Entries { - // Verify cached file still exists - if _, err := os.Stat(entry.CachePath); err == nil { - validEntries[path] = entry - } - } - - m.index.LoadFromMap(validEntries) - - log.Printf("[Cache] Loaded %d entries from index", len(validEntries)) - return nil -} - -// saveIndex persists the cache index to disk -func (m *Manager) saveIndex() error { - indexPath := m.config.CacheIndexPath() - - idx := &CacheIndex{ - Version: 1, - Entries: m.index.GetAllEntries(), - Stats: m.index.Stats(), - UpdatedAt: time.Now(), - } - - data, err := json.MarshalIndent(idx, "", " ") - if err != nil { - return err - } - - // Write atomically using temp file - tmpPath := indexPath + ".tmp" - if err := os.WriteFile(tmpPath, data, 0644); err != nil { - return err - } - - return os.Rename(tmpPath, indexPath) -} - -// detectContentType detects the content type based on file extension -func detectContentType(path string) string { - ext := strings.ToLower(filepath.Ext(path)) - - // Common media types - switch ext { - case ".mp4": - return "video/mp4" - case ".mkv": - return "video/x-matroska" - case ".avi": - return "video/x-msvideo" - case ".webm": - return "video/webm" - case ".mov": - return "video/quicktime" - case ".mp3": - return "audio/mpeg" - case ".flac": - return "audio/flac" - case ".wav": - return "audio/wav" - case ".aac": - return "audio/aac" - case ".jpg", ".jpeg": - return "image/jpeg" - case ".png": - return "image/png" - case ".gif": - return "image/gif" - case ".webp": - return "image/webp" - case ".srt": - return "text/plain" - case ".vtt": - return "text/vtt" - case ".ass", ".ssa": - return "text/x-ssa" - case ".nfo": - return "text/plain" - case ".txt": - return "text/plain" - case ".json": - return "application/json" - case ".xml": - return "application/xml" - default: - return "application/octet-stream" - } -} - -// ServeFile serves a file through the cache, handling HTTP response -func (m *Manager) ServeFile(w http.ResponseWriter, r *http.Request, path string) error { - cachedPath, contentType, err := m.Get(path) - if err != nil { - return err - } - - // Open the file (cached or original) - file, err := os.Open(cachedPath) - if err != nil { - return err - } - defer file.Close() - - stat, err := file.Stat() - if err != nil { - return err - } - - // Set content type if detected - if contentType != "" { - w.Header().Set("Content-Type", contentType) - } - - // Use http.ServeContent for proper range support - http.ServeContent(w, r, filepath.Base(path), stat.ModTime(), file) - return nil -} diff --git a/internal/cache/types.go b/internal/cache/types.go deleted file mode 100644 index 1f70886..0000000 --- a/internal/cache/types.go +++ /dev/null @@ -1,48 +0,0 @@ -package cache - -import ( - "time" -) - -// CacheEntry represents a cached file -type CacheEntry struct { - OriginalPath string `json:"originalPath"` // Relative to /files/ - CachePath string `json:"cachePath"` // Path in /meta-core/cache/ - Size int64 `json:"size"` - CachedAt time.Time `json:"cachedAt"` - LastAccessed time.Time `json:"lastAccessed"` - ContentType string `json:"contentType"` -} - -// CacheStats provides cache statistics -type CacheStats struct { - Enabled bool `json:"enabled"` - TotalSize int64 `json:"totalSize"` // Current cache size in bytes - MaxSize int64 `json:"maxSize"` // Maximum cache size in bytes - EntryCount int `json:"entryCount"` // Number of cached entries - HitCount int64 `json:"hitCount"` // Number of cache hits - MissCount int64 `json:"missCount"` // Number of cache misses - EvictionCount int64 `json:"evictionCount"` // Number of evictions - HitRate float64 `json:"hitRate"` // Hit rate percentage - TTLSeconds int `json:"ttlSeconds"` // TTL in seconds -} - -// CacheIndex is the persistent index structure -type CacheIndex struct { - Version int `json:"version"` - Entries map[string]*CacheEntry `json:"entries"` // keyed by OriginalPath - Stats *CacheStats `json:"stats"` - UpdatedAt time.Time `json:"updatedAt"` -} - -// NewCacheIndex creates a new empty cache index -func NewCacheIndex() *CacheIndex { - return &CacheIndex{ - Version: 1, - Entries: make(map[string]*CacheEntry), - Stats: &CacheStats{ - Enabled: true, - }, - UpdatedAt: time.Now(), - } -} diff --git a/internal/config/config.go b/internal/config/config.go index e4278e5..9b7e30f 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -39,12 +39,6 @@ type Config struct { // Mount configuration MountsDir string // Path to mounts configuration (default: /meta-core/mounts) - - // Cache configuration - CacheEnabled bool // Enable WebDAV caching (default: true) - CachePath string // Path to cache directory (default: /meta-core/cache) - CacheMaxSizeGB int // Maximum cache size in GB (default: 100) - CacheTTLSeconds int // Cache entry TTL in seconds (default: 3600) } // Load creates a Config from environment variables @@ -72,12 +66,6 @@ func Load() *Config { // Set mounts directory cfg.MountsDir = cfg.MetaCorePath + "/mounts" - // Set cache configuration - cfg.CacheEnabled = getEnvBool("CACHE_ENABLED", true) - cfg.CachePath = getEnv("CACHE_PATH", cfg.MetaCorePath+"/cache") - cfg.CacheMaxSizeGB = getEnvInt("CACHE_MAX_SIZE_GB", 100) - cfg.CacheTTLSeconds = getEnvInt("CACHE_TTL_SECONDS", 3600) - return cfg } @@ -111,16 +99,6 @@ func (c *Config) MountsErrorDir() string { return c.MountsDir + "/errors" } -// CacheDir returns the path to the cache directory -func (c *Config) CacheDir() string { - return c.CachePath -} - -// CacheIndexPath returns the path to the cache index file -func (c *Config) CacheIndexPath() string { - return c.CachePath + "/cache-index.json" -} - // WatchersFilePath returns the path to the watchers configuration file func (c *Config) WatchersFilePath() string { return c.MetaCorePath + "/watchers.json" diff --git a/internal/webdav/handler.go b/internal/webdav/handler.go index cde49e7..9f06c5b 100644 --- a/internal/webdav/handler.go +++ b/internal/webdav/handler.go @@ -10,25 +10,22 @@ import ( "strconv" "strings" - "github.com/metazla/meta-core/internal/cache" "github.com/metazla/meta-core/internal/config" "golang.org/x/net/webdav" ) -// Handler handles WebDAV requests with caching support +// Handler handles WebDAV requests (caching is handled by nginx proxy_cache) type Handler struct { - config *config.Config - cacheManager *cache.Manager - filesPath string - webdavFS webdav.Handler + config *config.Config + filesPath string + webdavFS webdav.Handler } -// NewHandler creates a new WebDAV handler with caching support -func NewHandler(cfg *config.Config, cacheManager *cache.Manager) *Handler { +// NewHandler creates a new WebDAV handler +func NewHandler(cfg *config.Config) *Handler { h := &Handler{ - config: cfg, - cacheManager: cacheManager, - filesPath: cfg.FilesPath, + config: cfg, + filesPath: cfg.FilesPath, } // Create standard WebDAV handler for write operations @@ -76,7 +73,7 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { } } -// handleRead handles GET and HEAD requests with caching +// handleRead handles GET and HEAD requests (caching is done by nginx proxy_cache) func (h *Handler) handleRead(w http.ResponseWriter, r *http.Request, path string) { fullPath := filepath.Join(h.filesPath, path) @@ -97,15 +94,8 @@ func (h *Handler) handleRead(w http.ResponseWriter, r *http.Request, path string return } - // Serve file through cache - if err := h.cacheManager.ServeFile(w, r, path); err != nil { - if os.IsNotExist(err) { - http.Error(w, "Not found", http.StatusNotFound) - } else { - log.Printf("[WebDAV] Error serving file: %v", err) - http.Error(w, err.Error(), http.StatusInternalServerError) - } - } + // Serve file directly (nginx handles caching via proxy_cache) + http.ServeFile(w, r, fullPath) } // handleDirListing returns a directory listing as JSON @@ -172,9 +162,7 @@ func (h *Handler) handlePut(w http.ResponseWriter, r *http.Request, path string) return } - // Invalidate cache for this path - h.cacheManager.Invalidate(path) - + // Note: nginx proxy_cache handles invalidation via TTL (1h inactive) w.WriteHeader(http.StatusCreated) } @@ -199,9 +187,7 @@ func (h *Handler) handleDelete(w http.ResponseWriter, r *http.Request, path stri return } - // Invalidate cache - h.cacheManager.Invalidate(path) - + // Note: nginx proxy_cache handles invalidation via TTL (1h inactive) w.WriteHeader(http.StatusNoContent) } @@ -268,10 +254,7 @@ func (h *Handler) handleCopyMove(w http.ResponseWriter, r *http.Request, path st http.Error(w, err.Error(), http.StatusInternalServerError) return } - - // Invalidate cache for both paths - h.cacheManager.Invalidate(path) - h.cacheManager.Invalidate(destPath) + // Note: nginx proxy_cache handles invalidation via TTL (1h inactive) } else { // Copy file if err := copyFile(srcFullPath, destFullPath); err != nil {