diff --git a/docker/nginx.conf b/docker/nginx.conf index 70f72d4..1373a4f 100644 --- a/docker/nginx.conf +++ b/docker/nginx.conf @@ -180,52 +180,51 @@ http { } # ========================================================================= - # WebDAV endpoint for file access - # Exposes /files directory for read/write operations + # WebDAV endpoint for file access (proxied to Go for caching) # Supports: GET (read), PUT (write), DELETE, MKCOL (mkdir), PROPFIND + # Files are cached by the Go service for faster repeated access # ========================================================================= location /webdav/ { - alias /files/; + proxy_pass http://meta-core-api/webdav/; + proxy_http_version 1.1; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; - # WebDAV methods - dav_methods PUT DELETE MKCOL COPY MOVE; - dav_access user:rw group:rw all:r; - - # Enable autoindex for directory listings (alternative to PROPFIND) - autoindex on; - autoindex_format json; - - # Allow creating directories automatically - create_full_put_path on; + # Disable buffering for streaming and large files + proxy_buffering off; + proxy_request_buffering off; # No size limit for uploads (large media files) client_max_body_size 0; - # Optimize for large file transfers - sendfile on; - tcp_nopush on; - tcp_nodelay on; - - # Output buffering for large files - output_buffers 1 512k; - aio threads; + # Timeout settings for large files + proxy_read_timeout 600s; + proxy_send_timeout 600s; # CORS headers for cross-container access 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, Range, If-Range' 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; # Handle OPTIONS preflight if ($request_method = 'OPTIONS') { add_header 'Access-Control-Allow-Origin' '*'; add_header 'Access-Control-Allow-Methods' 'GET, HEAD, PUT, DELETE, MKCOL, COPY, MOVE, PROPFIND, OPTIONS'; - add_header 'Access-Control-Allow-Headers' 'Authorization, Content-Type, Depth, Range, If-Range'; + add_header 'Access-Control-Allow-Headers' 'Authorization, Content-Type, Depth, Destination, Range, If-Range'; add_header 'Access-Control-Max-Age' 1728000; add_header 'Content-Type' 'text/plain; charset=utf-8'; add_header 'Content-Length' 0; return 204; } } + + location /webdav { + proxy_pass http://meta-core-api/webdav; + proxy_http_version 1.1; + proxy_set_header Host $host; + } } } diff --git a/go.mod b/go.mod index a06e419..fb5930f 100644 --- a/go.mod +++ b/go.mod @@ -6,10 +6,10 @@ require ( github.com/google/uuid v1.6.0 github.com/gorilla/mux v1.8.1 github.com/redis/go-redis/v9 v9.4.0 + golang.org/x/net v0.20.0 ) require ( github.com/cespare/xxhash/v2 v2.2.0 // indirect github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect - golang.org/x/sys v0.4.0 // indirect ) diff --git a/go.sum b/go.sum index be30923..4f6f7dd 100644 --- a/go.sum +++ b/go.sum @@ -12,5 +12,5 @@ github.com/gorilla/mux v1.8.1 h1:TuBL49tXwgrFYWhqrNgrUNEY92u81SPhu7sTdzQEiWY= github.com/gorilla/mux v1.8.1/go.mod h1:AKf9I4AEqPTmMytcMc0KkNouC66V3BtZ4qD5fmWSiMQ= github.com/redis/go-redis/v9 v9.4.0 h1:Yzoz33UZw9I/mFhx4MNrB6Fk+XHO1VukNcCa1+lwyKk= github.com/redis/go-redis/v9 v9.4.0/go.mod h1:hdY0cQFCN4fnSYT6TkisLufl/4W5UIXyv0b/CLO2V2M= -golang.org/x/sys v0.4.0 h1:Zr2JFtRQNX3BCZ8YtxRE9hNJYC8J6I1MVbMg6owUp18= -golang.org/x/sys v0.4.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/net v0.20.0 h1:aCL9BSgETF1k+blQaYUBx9hJ9LOGP3gAVemcZlf1Kpo= +golang.org/x/net v0.20.0/go.mod h1:z8BVo6PvndSri0LbOE3hAn0apkU+1YvI6E70E9jsnvY= diff --git a/internal/api/server.go b/internal/api/server.go index a26b46e..c2e7895 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -8,13 +8,16 @@ 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" "github.com/metazla/meta-core/internal/leader" "github.com/metazla/meta-core/internal/mounts" "github.com/metazla/meta-core/internal/storage" "github.com/metazla/meta-core/internal/watcher" "github.com/metazla/meta-core/internal/watchers" + "github.com/metazla/meta-core/internal/webdav" ) // Server is the HTTP API server for meta-core @@ -33,6 +36,11 @@ 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 server *http.Server } @@ -100,6 +108,18 @@ 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 + } + s.setupRoutes() return s } @@ -179,6 +199,19 @@ 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) + if s.webdavHandler != nil { + s.router.PathPrefix("/webdav/").Handler(s.webdavHandler) + s.router.PathPrefix("/webdav").Handler(s.webdavHandler) + log.Println("[API] WebDAV routes registered at /webdav/*") + } + // Add middleware s.router.Use(loggingMiddleware) s.router.Use(corsMiddleware) @@ -218,6 +251,30 @@ 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() { + s.metaPublisher = events.NewMetaPublisher(s.storage.GetRedisClient()) + if err := s.metaPublisher.Start(); err != nil { + log.Printf("[API] Warning: failed to start meta publisher: %v", err) + } + } + return nil } @@ -237,6 +294,27 @@ func (s *Server) Stop() error { } } + // Stop meta publisher + if s.metaPublisher != nil { + if err := s.metaPublisher.Stop(); err != nil { + log.Printf("[API] Warning: failed to stop meta publisher: %v", err) + } + } + + // 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 new file mode 100644 index 0000000..af8404a --- /dev/null +++ b/internal/cache/handlers.go @@ -0,0 +1,73 @@ +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 new file mode 100644 index 0000000..1970d7e --- /dev/null +++ b/internal/cache/index.go @@ -0,0 +1,242 @@ +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 new file mode 100644 index 0000000..edcad6c --- /dev/null +++ b/internal/cache/invalidator.go @@ -0,0 +1,166 @@ +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 new file mode 100644 index 0000000..85c027d --- /dev/null +++ b/internal/cache/manager.go @@ -0,0 +1,494 @@ +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 new file mode 100644 index 0000000..1f70886 --- /dev/null +++ b/internal/cache/types.go @@ -0,0 +1,48 @@ +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 9b7e30f..e4278e5 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -39,6 +39,12 @@ 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 @@ -66,6 +72,12 @@ 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 } @@ -99,6 +111,16 @@ 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/events/meta_publisher.go b/internal/events/meta_publisher.go new file mode 100644 index 0000000..1f05a38 --- /dev/null +++ b/internal/events/meta_publisher.go @@ -0,0 +1,139 @@ +package events + +import ( + "context" + "fmt" + "log" + "strings" + "sync" + "time" + + "github.com/redis/go-redis/v9" +) + +const ( + // MetaEventsStream is the Redis stream name for metadata events + MetaEventsStream = "meta:events" + // StreamMaxLen is the maximum stream length (approximate) + StreamMaxLen = 10000 +) + +// MetaPublisher subscribes to Redis keyspace notifications +// and publishes metadata change events to the meta:events stream +type MetaPublisher struct { + client *redis.Client + ctx context.Context + cancel context.CancelFunc + wg sync.WaitGroup + running bool + mu sync.RWMutex +} + +// NewMetaPublisher creates a new MetaPublisher +func NewMetaPublisher(client *redis.Client) *MetaPublisher { + ctx, cancel := context.WithCancel(context.Background()) + return &MetaPublisher{ + client: client, + ctx: ctx, + cancel: cancel, + } +} + +// Start begins subscribing to keyspace notifications +func (p *MetaPublisher) Start() error { + p.mu.Lock() + if p.running { + p.mu.Unlock() + return nil + } + p.running = true + p.mu.Unlock() + + p.wg.Add(1) + go p.subscribeLoop() + + log.Println("[MetaPublisher] Started - listening for keyspace notifications") + return nil +} + +// Stop stops the publisher +func (p *MetaPublisher) Stop() error { + p.mu.Lock() + if !p.running { + p.mu.Unlock() + return nil + } + p.running = false + p.mu.Unlock() + + p.cancel() + p.wg.Wait() + + log.Println("[MetaPublisher] Stopped") + return nil +} + +// IsRunning returns true if the publisher is running +func (p *MetaPublisher) IsRunning() bool { + p.mu.RLock() + defer p.mu.RUnlock() + return p.running +} + +// subscribeLoop subscribes to keyspace notifications and publishes events +func (p *MetaPublisher) subscribeLoop() { + defer p.wg.Done() + + // Subscribe to all file:* keyspace events + // Pattern: __keyspace@0__:file:* + pattern := "__keyspace@0__:file:*" + pubsub := p.client.PSubscribe(p.ctx, pattern) + defer pubsub.Close() + + log.Printf("[MetaPublisher] Subscribed to pattern: %s", pattern) + + ch := pubsub.Channel() + for { + select { + case <-p.ctx.Done(): + return + case msg := <-ch: + if msg == nil { + continue + } + // msg.Channel: __keyspace@0__:file:midhash256:abc123/tmdb + // msg.Payload: set, del, expire, etc. + p.publishEvent(msg.Channel, msg.Payload) + } + } +} + +// publishEvent publishes a keyspace event to the meta:events stream +func (p *MetaPublisher) publishEvent(channel, operation string) { + // Extract key from channel: __keyspace@0__:file:xxx/field -> file:xxx/field + key := strings.TrimPrefix(channel, "__keyspace@0__:") + + // Skip index key and non-file keys + if strings.Contains(key, "__index__") || !strings.HasPrefix(key, "file:") { + return + } + + // Build event fields + fields := map[string]interface{}{ + "type": operation, // "set", "del", "expire" + "key": key, // "file:midhash256:abc123/tmdb" + "ts": fmt.Sprintf("%d", time.Now().UnixMilli()), + } + + // Add to stream + _, err := p.client.XAdd(p.ctx, &redis.XAddArgs{ + Stream: MetaEventsStream, + MaxLen: StreamMaxLen, + Approx: true, + Values: fields, + }).Result() + + if err != nil { + log.Printf("[MetaPublisher] Error publishing event: %v", err) + } +} diff --git a/internal/leader/redis.go b/internal/leader/redis.go index 5ac7fb3..785bd8b 100644 --- a/internal/leader/redis.go +++ b/internal/leader/redis.go @@ -57,6 +57,11 @@ func (rm *RedisManager) Start() error { "--dbfilename", "dump.rdb", "--save", "60", "1", // Save after 60 seconds if at least 1 key changed "--loglevel", "warning", + // Enable keyspace notifications for meta:events stream + // K = Keyspace events (publish to __keyspace@0__:keyname) + // g = Generic commands (DEL, EXPIRE, RENAME) + // $ = String commands (SET, MSET, etc.) + "--notify-keyspace-events", "Kg$", } rm.cmd = exec.Command("redis-server", args...) diff --git a/internal/storage/client.go b/internal/storage/client.go index 5ae8ec4..3b08780 100644 --- a/internal/storage/client.go +++ b/internal/storage/client.go @@ -97,9 +97,9 @@ func (c *Client) buildKey(key string) string { return key } -// buildHashKey constructs the Redis Hash key for a file -func (c *Client) buildHashKey(hashID string) string { - return c.buildKey(fmt.Sprintf("file:%s", hashID)) +// buildKeyPrefix constructs the prefix for a file's flat keys +func (c *Client) buildKeyPrefix(hashID string) string { + return c.buildKey(fmt.Sprintf("file:%s/", hashID)) } // buildIndexKey constructs the key for the file index set @@ -108,7 +108,7 @@ func (c *Client) buildIndexKey() string { } // GetMetadataFlat retrieves all metadata for a file as a flat map -// Uses Redis Hash: HGETALL file:{hashId} +// Uses flat keys: SCAN file:{hashId}/* then MGET func (c *Client) GetMetadataFlat(hashID string) (map[string]string, error) { c.mu.RLock() defer c.mu.RUnlock() @@ -120,10 +120,34 @@ func (c *Client) GetMetadataFlat(hashID string) (map[string]string, error) { ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() - hashKey := c.buildHashKey(hashID) - result, err := c.client.HGetAll(ctx, hashKey).Result() - if err != nil { - return nil, fmt.Errorf("hgetall failed: %w", err) + prefix := c.buildKeyPrefix(hashID) + result := make(map[string]string) + + // Scan for all keys with this prefix + var cursor uint64 + for { + keys, nextCursor, err := c.client.Scan(ctx, cursor, prefix+"*", 1000).Result() + if err != nil { + return nil, fmt.Errorf("scan failed: %w", err) + } + + if len(keys) > 0 { + values, err := c.client.MGet(ctx, keys...).Result() + if err != nil { + return nil, fmt.Errorf("mget failed: %w", err) + } + for i, key := range keys { + field := strings.TrimPrefix(key, prefix) + if values[i] != nil { + result[field] = values[i].(string) + } + } + } + + cursor = nextCursor + if cursor == 0 { + break + } } if len(result) == 0 { @@ -133,8 +157,8 @@ func (c *Client) GetMetadataFlat(hashID string) (map[string]string, error) { return result, nil } -// SetMetadataFlat stores metadata for a file using Redis Hash -// Uses Redis Hash: HMSET file:{hashId} prop1 val1 prop2 val2... +// SetMetadataFlat stores metadata for a file using flat keys +// Uses pipeline SET for each key: file:{hashId}/{property} func (c *Client) SetMetadataFlat(hashID string, metadata map[string]string) error { c.mu.RLock() defer c.mu.RUnlock() @@ -150,11 +174,15 @@ func (c *Client) SetMetadataFlat(hashID string, metadata map[string]string) erro ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() - hashKey := c.buildHashKey(hashID) + prefix := c.buildKeyPrefix(hashID) + pipe := c.client.Pipeline() - // Use HMSET to set all fields at once - if err := c.client.HMSet(ctx, hashKey, metadata).Err(); err != nil { - return fmt.Errorf("hmset failed: %w", err) + for field, value := range metadata { + pipe.Set(ctx, prefix+field, value, 0) + } + + if _, err := pipe.Exec(ctx); err != nil { + return fmt.Errorf("pipeline exec failed: %w", err) } // Add to index set @@ -188,7 +216,7 @@ func (c *Client) GetAllHashIDs() ([]string, error) { } // GetProperty retrieves a single property value -// Uses Redis Hash: HGET file:{hashId} {property} +// Uses flat key: GET file:{hashId}/{property} func (c *Client) GetProperty(hashID, property string) (string, error) { c.mu.RLock() defer c.mu.RUnlock() @@ -200,8 +228,8 @@ func (c *Client) GetProperty(hashID, property string) (string, error) { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - hashKey := c.buildHashKey(hashID) - result, err := c.client.HGet(ctx, hashKey, property).Result() + key := c.buildKeyPrefix(hashID) + property + result, err := c.client.Get(ctx, key).Result() if err == redis.Nil { return "", nil } @@ -209,7 +237,7 @@ func (c *Client) GetProperty(hashID, property string) (string, error) { } // SetProperty sets a single property value -// Uses Redis Hash: HSET file:{hashId} {property} {value} +// Uses flat key: SET file:{hashId}/{property} {value} func (c *Client) SetProperty(hashID, property, value string) error { c.mu.RLock() defer c.mu.RUnlock() @@ -221,11 +249,11 @@ func (c *Client) SetProperty(hashID, property, value string) error { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - hashKey := c.buildHashKey(hashID) + key := c.buildKeyPrefix(hashID) + property - // Use HSET to set the field - if err := c.client.HSet(ctx, hashKey, property, value).Err(); err != nil { - return fmt.Errorf("hset failed: %w", err) + // Use SET to set the key + if err := c.client.Set(ctx, key, value, 0).Err(); err != nil { + return fmt.Errorf("set failed: %w", err) } // Add to index set @@ -237,7 +265,7 @@ func (c *Client) SetProperty(hashID, property, value string) error { } // DeleteMetadata deletes all metadata for a file -// Uses Redis Hash: DEL file:{hashId} +// Uses SCAN + DEL for flat keys: file:{hashId}/* func (c *Client) DeleteMetadata(hashID string) (int64, error) { c.mu.RLock() defer c.mu.RUnlock() @@ -249,25 +277,37 @@ func (c *Client) DeleteMetadata(hashID string) (int64, error) { ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() - hashKey := c.buildHashKey(hashID) + prefix := c.buildKeyPrefix(hashID) - // Get field count before deletion - fieldCount, err := c.client.HLen(ctx, hashKey).Result() - if err != nil { - return 0, fmt.Errorf("hlen failed: %w", err) - } + // Scan and delete all keys with this prefix + var deletedCount int64 + var cursor uint64 + for { + keys, nextCursor, err := c.client.Scan(ctx, cursor, prefix+"*", 1000).Result() + if err != nil { + return deletedCount, fmt.Errorf("scan failed: %w", err) + } - // Delete the hash - if err := c.client.Del(ctx, hashKey).Err(); err != nil { - return 0, fmt.Errorf("del failed: %w", err) + if len(keys) > 0 { + deleted, err := c.client.Del(ctx, keys...).Result() + if err != nil { + return deletedCount, fmt.Errorf("del failed: %w", err) + } + deletedCount += deleted + } + + cursor = nextCursor + if cursor == 0 { + break + } } // Remove from index set if err := c.client.SRem(ctx, c.buildIndexKey(), hashID).Err(); err != nil { - return fieldCount, fmt.Errorf("srem index failed: %w", err) + return deletedCount, fmt.Errorf("srem index failed: %w", err) } - return fieldCount, nil + return deletedCount, nil } // CountFiles returns the number of unique files stored @@ -302,23 +342,23 @@ func (c *Client) LookupPathByCID(cid string) (string, error) { return "", err } - // For each hash, check poster and backdrop CIDs using Hash operations + // For each hash, check poster and backdrop CIDs using flat keys for _, hashID := range hashIDs { - hashKey := c.buildHashKey(hashID) + prefix := c.buildKeyPrefix(hashID) // Check poster - posterCID, err := c.client.HGet(ctx, hashKey, "poster").Result() + posterCID, err := c.client.Get(ctx, prefix+"poster").Result() if err == nil && posterCID == cid { - posterPath, err := c.client.HGet(ctx, hashKey, "posterPath").Result() + posterPath, err := c.client.Get(ctx, prefix+"posterPath").Result() if err == nil && posterPath != "" { return posterPath, nil } } // Check backdrop - backdropCID, err := c.client.HGet(ctx, hashKey, "backdrop").Result() + backdropCID, err := c.client.Get(ctx, prefix+"backdrop").Result() if err == nil && backdropCID == cid { - backdropPath, err := c.client.HGet(ctx, hashKey, "backdropPath").Result() + backdropPath, err := c.client.Get(ctx, prefix+"backdropPath").Result() if err == nil && backdropPath != "" { return backdropPath, nil } @@ -341,7 +381,7 @@ func (c *Client) getAllHashIDsInternal(ctx context.Context) ([]string, error) { // MergeMetadataFlat merges new metadata into existing (PATCH semantics) // New keys are added, existing keys are updated, missing keys are NOT deleted -// Uses Redis Hash: HMSET file:{hashId} +// Uses pipeline SET for flat keys: file:{hashId}/{property} func (c *Client) MergeMetadataFlat(hashID string, metadata map[string]string) (int, error) { c.mu.RLock() defer c.mu.RUnlock() @@ -357,11 +397,15 @@ func (c *Client) MergeMetadataFlat(hashID string, metadata map[string]string) (i ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() - hashKey := c.buildHashKey(hashID) + prefix := c.buildKeyPrefix(hashID) + pipe := c.client.Pipeline() - // Use HMSET to merge fields - if err := c.client.HMSet(ctx, hashKey, metadata).Err(); err != nil { - return 0, fmt.Errorf("hmset failed: %w", err) + for field, value := range metadata { + pipe.Set(ctx, prefix+field, value, 0) + } + + if _, err := pipe.Exec(ctx); err != nil { + return 0, fmt.Errorf("pipeline exec failed: %w", err) } // Add to index set @@ -373,7 +417,7 @@ func (c *Client) MergeMetadataFlat(hashID string, metadata map[string]string) (i } // DeleteProperty deletes a single property -// Uses Redis Hash: HDEL file:{hashId} {property} +// Uses flat key: DEL file:{hashId}/{property} func (c *Client) DeleteProperty(hashID, property string) error { c.mu.RLock() defer c.mu.RUnlock() @@ -385,11 +429,11 @@ func (c *Client) DeleteProperty(hashID, property string) error { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - hashKey := c.buildHashKey(hashID) - return c.client.HDel(ctx, hashKey, property).Err() + key := c.buildKeyPrefix(hashID) + property + return c.client.Del(ctx, key).Err() } -// AddToSet adds a value to a set-type field (stored as pipe-delimited string in Hash field) +// AddToSet adds a value to a set-type field (stored as pipe-delimited string in flat key) // Returns true if value was added, false if it already existed func (c *Client) AddToSet(hashID, property, value string) (bool, error) { c.mu.Lock() @@ -402,10 +446,10 @@ func (c *Client) AddToSet(hashID, property, value string) (bool, error) { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - hashKey := c.buildHashKey(hashID) + key := c.buildKeyPrefix(hashID) + property // Get current value - current, err := c.client.HGet(ctx, hashKey, property).Result() + current, err := c.client.Get(ctx, key).Result() if err != nil && err != redis.Nil { return false, err } @@ -427,8 +471,8 @@ func (c *Client) AddToSet(hashID, property, value string) (bool, error) { values = append(values, value) newValue := strings.Join(values, "|") - // Save back using HSET - if err := c.client.HSet(ctx, hashKey, property, newValue).Err(); err != nil { + // Save back using SET + if err := c.client.Set(ctx, key, newValue, 0).Err(); err != nil { return false, err } @@ -440,7 +484,7 @@ func (c *Client) AddToSet(hashID, property, value string) (bool, error) { return true, nil } -// RemoveFromSet removes a value from a set-type field (stored in Hash) +// RemoveFromSet removes a value from a set-type field (stored in flat key) // Returns true if value was removed, false if it didn't exist func (c *Client) RemoveFromSet(hashID, property, value string) (bool, error) { c.mu.Lock() @@ -453,12 +497,12 @@ func (c *Client) RemoveFromSet(hashID, property, value string) (bool, error) { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - hashKey := c.buildHashKey(hashID) + key := c.buildKeyPrefix(hashID) + property // Get current value - current, err := c.client.HGet(ctx, hashKey, property).Result() + current, err := c.client.Get(ctx, key).Result() if err == redis.Nil { - return false, nil // Field doesn't exist + return false, nil // Key doesn't exist } if err != nil { return false, err @@ -482,13 +526,13 @@ func (c *Client) RemoveFromSet(hashID, property, value string) (bool, error) { return false, nil } - // Save back (or delete field if empty) + // Save back (or delete key if empty) if len(newValues) == 0 { - return true, c.client.HDel(ctx, hashKey, property).Err() + return true, c.client.Del(ctx, key).Err() } newValue := strings.Join(newValues, "|") - return true, c.client.HSet(ctx, hashKey, property, newValue).Err() + return true, c.client.Set(ctx, key, newValue, 0).Err() } // GetMemoryInfo returns Redis memory usage information @@ -540,12 +584,29 @@ func (c *Client) ClearAllMetadata() (int64, error) { var deletedCount int64 - // Delete each file's hash + // Delete each file's flat keys for _, hashID := range hashIDs { - hashKey := c.buildHashKey(hashID) - if err := c.client.Del(ctx, hashKey).Err(); err != nil { - log.Printf("[Storage] Warning: failed to delete %s: %v", hashKey, err) - continue + prefix := c.buildKeyPrefix(hashID) + + // Scan and delete all keys with this prefix + var cursor uint64 + for { + keys, nextCursor, err := c.client.Scan(ctx, cursor, prefix+"*", 1000).Result() + if err != nil { + log.Printf("[Storage] Warning: failed to scan %s: %v", prefix, err) + break + } + + if len(keys) > 0 { + if err := c.client.Del(ctx, keys...).Err(); err != nil { + log.Printf("[Storage] Warning: failed to delete keys for %s: %v", hashID, err) + } + } + + cursor = nextCursor + if cursor == 0 { + break + } } deletedCount++ } @@ -599,3 +660,42 @@ func (c *Client) XAdd(stream string, maxLen int64, fields map[string]interface{} return id, nil } + +// ClearStream deletes a Redis stream entirely +func (c *Client) ClearStream(stream string) error { + c.mu.RLock() + defer c.mu.RUnlock() + + if c.client == nil { + return fmt.Errorf("not connected") + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + if err := c.client.Del(ctx, stream).Err(); err != nil { + return fmt.Errorf("del stream failed: %w", err) + } + + return nil +} + +// GetStreamLength returns the length of a Redis stream +func (c *Client) GetStreamLength(stream string) (int64, error) { + c.mu.RLock() + defer c.mu.RUnlock() + + if c.client == nil { + return 0, fmt.Errorf("not connected") + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + length, err := c.client.XLen(ctx, stream).Result() + if err != nil { + return 0, fmt.Errorf("xlen failed: %w", err) + } + + return length, nil +} diff --git a/internal/watcher/dispatcher.go b/internal/watcher/dispatcher.go index 8858aa2..c9bb4b0 100644 --- a/internal/watcher/dispatcher.go +++ b/internal/watcher/dispatcher.go @@ -66,6 +66,9 @@ func (d *Dispatcher) dispatchToStream(event FileEvent) { if event.OldPath != "" { fields["oldPath"] = event.OldPath } + if event.WatcherID != "" { + fields["watcherId"] = event.WatcherID + } // Publish to stream _, err := client.XAdd(EventsStream, StreamMaxLen, fields) @@ -73,3 +76,49 @@ func (d *Dispatcher) dispatchToStream(event FileEvent) { log.Printf("[Dispatcher] Failed to publish to stream: %v", err) } } + +// EmitReset clears the stream and emits a reset event +// This is called when triggering a full rescan +func (d *Dispatcher) EmitReset(watcherID string) error { + d.mu.RLock() + client := d.storageClient + d.mu.RUnlock() + + if client == nil || !client.IsConnected() { + return fmt.Errorf("storage client not connected") + } + + // Delete the stream to clear all historical events + if err := client.ClearStream(EventsStream); err != nil { + log.Printf("[Dispatcher] Warning: failed to clear stream: %v", err) + // Continue anyway - we'll emit the reset event + } + + log.Printf("[Dispatcher] Cleared stream %s for reset", EventsStream) + + // Emit reset event as the first message in the fresh stream + event := FileEvent{ + Type: EventTypeReset, + Path: "", // No specific path for reset + WatcherID: watcherID, + Timestamp: NowMS(), + } + + // Dispatch synchronously for reset + d.dispatchToStream(event) + + return nil +} + +// GetStreamLength returns the current stream length +func (d *Dispatcher) GetStreamLength() (int64, error) { + d.mu.RLock() + client := d.storageClient + d.mu.RUnlock() + + if client == nil || !client.IsConnected() { + return 0, fmt.Errorf("storage client not connected") + } + + return client.GetStreamLength(EventsStream) +} diff --git a/internal/watcher/state.go b/internal/watcher/state.go new file mode 100644 index 0000000..d12069e --- /dev/null +++ b/internal/watcher/state.go @@ -0,0 +1,146 @@ +package watcher + +import ( + "sync" +) + +// FileState tracks the known state of a file +type FileState struct { + Size int64 + MtimeNano int64 + MidHash string +} + +// WatcherState tracks all files for a single watcher (in-memory) +type WatcherState struct { + mu sync.RWMutex + files map[string]*FileState // key: relative path +} + +// NewWatcherState creates a new watcher state tracker +func NewWatcherState() *WatcherState { + return &WatcherState{ + files: make(map[string]*FileState), + } +} + +// Get returns the file state for a given path, or nil if not tracked +func (ws *WatcherState) Get(path string) *FileState { + ws.mu.RLock() + defer ws.mu.RUnlock() + return ws.files[path] +} + +// Set updates or adds the file state for a given path +func (ws *WatcherState) Set(path string, state *FileState) { + ws.mu.Lock() + defer ws.mu.Unlock() + ws.files[path] = state +} + +// Delete removes a file from the tracked state +func (ws *WatcherState) Delete(path string) { + ws.mu.Lock() + defer ws.mu.Unlock() + delete(ws.files, path) +} + +// GetAll returns a snapshot of all tracked files +// The returned map is a copy - safe to modify +func (ws *WatcherState) GetAll() map[string]*FileState { + ws.mu.RLock() + defer ws.mu.RUnlock() + + snapshot := make(map[string]*FileState, len(ws.files)) + for path, state := range ws.files { + // Copy the state to avoid race conditions + stateCopy := *state + snapshot[path] = &stateCopy + } + return snapshot +} + +// Clear removes all tracked files +func (ws *WatcherState) Clear() { + ws.mu.Lock() + defer ws.mu.Unlock() + ws.files = make(map[string]*FileState) +} + +// Count returns the number of tracked files +func (ws *WatcherState) Count() int { + ws.mu.RLock() + defer ws.mu.RUnlock() + return len(ws.files) +} + +// StateRegistry manages per-watcher state +type StateRegistry struct { + mu sync.RWMutex + states map[string]*WatcherState // key: watcher ID +} + +// NewStateRegistry creates a new state registry +func NewStateRegistry() *StateRegistry { + return &StateRegistry{ + states: make(map[string]*WatcherState), + } +} + +// GetOrCreate returns the state for a watcher, creating one if needed +func (sr *StateRegistry) GetOrCreate(watcherID string) *WatcherState { + sr.mu.Lock() + defer sr.mu.Unlock() + + state, exists := sr.states[watcherID] + if !exists { + state = NewWatcherState() + sr.states[watcherID] = state + } + return state +} + +// Get returns the state for a watcher, or nil if not exists +func (sr *StateRegistry) Get(watcherID string) *WatcherState { + sr.mu.RLock() + defer sr.mu.RUnlock() + return sr.states[watcherID] +} + +// Clear removes the state for a watcher +func (sr *StateRegistry) Clear(watcherID string) { + sr.mu.Lock() + defer sr.mu.Unlock() + + if state, exists := sr.states[watcherID]; exists { + state.Clear() + } +} + +// ClearAll removes all watcher states +func (sr *StateRegistry) ClearAll() { + sr.mu.Lock() + defer sr.mu.Unlock() + + for _, state := range sr.states { + state.Clear() + } +} + +// Delete removes a watcher's state completely +func (sr *StateRegistry) Delete(watcherID string) { + sr.mu.Lock() + defer sr.mu.Unlock() + delete(sr.states, watcherID) +} + +// Count returns the number of files tracked for a watcher +func (sr *StateRegistry) Count(watcherID string) int { + sr.mu.RLock() + defer sr.mu.RUnlock() + + if state, exists := sr.states[watcherID]; exists { + return state.Count() + } + return 0 +} diff --git a/internal/watcher/types.go b/internal/watcher/types.go index 0a0820f..711d256 100644 --- a/internal/watcher/types.go +++ b/internal/watcher/types.go @@ -10,6 +10,7 @@ const ( EventTypeChange FileEventType = "change" EventTypeDelete FileEventType = "delete" EventTypeRename FileEventType = "rename" + EventTypeReset FileEventType = "reset" ) // FileEvent represents a file system event @@ -20,6 +21,7 @@ type FileEvent struct { Timestamp int64 `json:"timestamp"` MidHash256 string `json:"midhash256,omitempty"` // midhash256 CID (SHA-256 of middle 1MB + file size) OldPath string `json:"oldPath,omitempty"` // For rename events + WatcherID string `json:"watcherId,omitempty"` // For reset events } // PendingEvent tracks a file event that's being debounced @@ -44,6 +46,14 @@ type ScanStatusResponse struct { FileCount int `json:"fileCount,omitempty"` } +// DiffScanResult holds the results of a differential scan +type DiffScanResult struct { + Added int // Number of new files added + Changed int // Number of files with changed size/mtime + Deleted int // Number of files that no longer exist + Total int // Total number of files currently tracked +} + // NowMS returns current time in milliseconds func NowMS() int64 { return time.Now().UnixMilli() diff --git a/internal/watcher/watcher.go b/internal/watcher/watcher.go index 084388c..f13ce64 100644 --- a/internal/watcher/watcher.go +++ b/internal/watcher/watcher.go @@ -13,10 +13,11 @@ import ( // Watcher provides file scanning and event dispatching type Watcher struct { - config *config.Config - debouncer *Debouncer - dispatcher *Dispatcher - filesPath string + config *config.Config + debouncer *Debouncer + dispatcher *Dispatcher + filesPath string + stateRegistry *StateRegistry mu sync.RWMutex eventBuffer []FileEvent @@ -27,11 +28,12 @@ func NewWatcher(cfg *config.Config, dispatcher *Dispatcher) (*Watcher, error) { debouncer := NewDebouncer(0) // No debouncing for polling-based scanning w := &Watcher{ - config: cfg, - debouncer: debouncer, - dispatcher: dispatcher, - filesPath: cfg.FilesPath, - eventBuffer: make([]FileEvent, 0, 1000), + config: cfg, + debouncer: debouncer, + dispatcher: dispatcher, + filesPath: cfg.FilesPath, + stateRegistry: NewStateRegistry(), + eventBuffer: make([]FileEvent, 0, 1000), } return w, nil @@ -110,6 +112,138 @@ func (w *Watcher) ScanDirectory(root string) int { return count } +// ScanDirectoryDiff scans a directory and only emits events for differences +// Returns the count of added, changed, and deleted files +func (w *Watcher) ScanDirectoryDiff(root string, watcherID string) DiffScanResult { + result := DiffScanResult{} + state := w.stateRegistry.GetOrCreate(watcherID) + + // Get snapshot of known files + knownFiles := state.GetAll() + seenFiles := make(map[string]bool) + + filepath.Walk(root, func(path string, info os.FileInfo, err error) error { + if err != nil { + return nil + } + + // Skip directories + if info.IsDir() { + return nil + } + + // Skip hidden files + if strings.HasPrefix(filepath.Base(path), ".") { + return nil + } + + // Get relative path + relPath, err := filepath.Rel(w.filesPath, path) + if err != nil { + relPath = path + } + relPath = filepath.ToSlash(relPath) + seenFiles[relPath] = true + + // Check current file state + currentMtime := info.ModTime().UnixNano() + currentSize := info.Size() + + prevState := knownFiles[relPath] + + if prevState == nil { + // New file - emit add event + event := FileEvent{ + Type: EventTypeAdd, + Path: relPath, + Size: currentSize, + Timestamp: NowMS(), + } + + // Compute midhash256 + if hash, err := ComputeMidHash256(path); err == nil { + event.MidHash256 = hash + } + + // Dispatch event + w.dispatcher.Dispatch(event) + + // Update state + state.Set(relPath, &FileState{ + Size: currentSize, + MtimeNano: currentMtime, + MidHash: event.MidHash256, + }) + + result.Added++ + } else if prevState.Size != currentSize || prevState.MtimeNano != currentMtime { + // File changed - emit change event + event := FileEvent{ + Type: EventTypeChange, + Path: relPath, + Size: currentSize, + Timestamp: NowMS(), + } + + // Compute midhash256 + if hash, err := ComputeMidHash256(path); err == nil { + event.MidHash256 = hash + } + + // Dispatch event + w.dispatcher.Dispatch(event) + + // Update state + state.Set(relPath, &FileState{ + Size: currentSize, + MtimeNano: currentMtime, + MidHash: event.MidHash256, + }) + + result.Changed++ + } + // If file exists and hasn't changed, do nothing + + return nil + }) + + // Check for deleted files (in known but not seen) + for relPath := range knownFiles { + if !seenFiles[relPath] { + // File deleted - emit delete event + event := FileEvent{ + Type: EventTypeDelete, + Path: relPath, + Timestamp: NowMS(), + } + + w.dispatcher.Dispatch(event) + state.Delete(relPath) + result.Deleted++ + } + } + + result.Total = state.Count() + return result +} + +// ClearState clears the tracked state for a watcher +func (w *Watcher) ClearState(watcherID string) { + w.stateRegistry.Clear(watcherID) +} + +// ClearEventBuffer clears the in-memory event buffer +func (w *Watcher) ClearEventBuffer() { + w.mu.Lock() + defer w.mu.Unlock() + w.eventBuffer = make([]FileEvent, 0, 1000) +} + +// GetStateRegistry returns the state registry for external access +func (w *Watcher) GetStateRegistry() *StateRegistry { + return w.stateRegistry +} + // ScanMountPath scans a specific mount path and returns file count // This is used by the mount poller to scan individual mounts func (w *Watcher) ScanMountPath(mountPath string) (int, error) { diff --git a/internal/watchers/handlers.go b/internal/watchers/handlers.go index 4c59439..32fe44e 100644 --- a/internal/watchers/handlers.go +++ b/internal/watchers/handlers.go @@ -26,10 +26,12 @@ func (h *Handlers) RegisterRoutes(r *mux.Router) { r.HandleFunc("/api/watchers", h.handleList).Methods("GET") r.HandleFunc("/api/watchers", h.handleCreate).Methods("POST") r.HandleFunc("/api/watchers/scan-all", h.handleScanAll).Methods("POST") + r.HandleFunc("/api/watchers/reset-all", h.handleResetAll).Methods("POST") r.HandleFunc("/api/watchers/{id}", h.handleGet).Methods("GET") r.HandleFunc("/api/watchers/{id}", h.handleUpdate).Methods("PUT") r.HandleFunc("/api/watchers/{id}", h.handleDelete).Methods("DELETE") r.HandleFunc("/api/watchers/{id}/scan", h.handleScan).Methods("POST") + r.HandleFunc("/api/watchers/{id}/reset", h.handleReset).Methods("POST") } // handleList handles GET /api/watchers @@ -176,6 +178,44 @@ func (h *Handlers) handleScanAll(w http.ResponseWriter, r *http.Request) { }) } +// handleReset handles POST /api/watchers/{id}/reset +func (h *Handlers) handleReset(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + id := vars["id"] + + fileCount, err := h.poller.TriggerReset(id) + if err != nil { + if err.Error() == "watcher not found: "+id { + writeError(w, http.StatusNotFound, err.Error()) + } else { + writeError(w, http.StatusInternalServerError, err.Error()) + } + return + } + + writeJSON(w, http.StatusOK, map[string]interface{}{ + "status": "ok", + "message": "Reset complete", + "watcherId": id, + "fileCount": fileCount, + }) +} + +// handleResetAll handles POST /api/watchers/reset-all +func (h *Handlers) handleResetAll(w http.ResponseWriter, r *http.Request) { + totalCount, err := h.poller.TriggerResetAll() + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + + writeJSON(w, http.StatusOK, map[string]interface{}{ + "status": "ok", + "message": "All watchers reset", + "totalFiles": totalCount, + }) +} + // Helper functions func writeJSON(w http.ResponseWriter, status int, data interface{}) { diff --git a/internal/watchers/poller.go b/internal/watchers/poller.go index e767a89..f02fbf4 100644 --- a/internal/watchers/poller.go +++ b/internal/watchers/poller.go @@ -203,7 +203,7 @@ func (p *Poller) stopPoller(poller *watcherPoller) { } } -// runPollScan executes a scan for a watcher +// runPollScan executes a differential scan for a watcher func (p *Poller) runPollScan(watcherID, path string) { if p.scanner == nil { return @@ -216,22 +216,26 @@ func (p *Poller) runPollScan(watcherID, path string) { } p.mu.Unlock() - // Perform scan - fileCount := p.scanner.ScanDirectory(path) + // Perform differential scan + result := p.scanner.ScanDirectoryDiff(path, watcherID) // Update status p.mu.Lock() if status, exists := p.status[watcherID]; exists { status.LastScan = NowMS() - status.FileCount = fileCount + status.FileCount = result.Total status.IsScanning = false } p.mu.Unlock() - log.Printf("[WatcherPoller] Poll scan complete for watcher %s: %d files", watcherID, fileCount) + // Only log if there were changes + if result.Added > 0 || result.Changed > 0 || result.Deleted > 0 { + log.Printf("[WatcherPoller] Diff scan for %s: +%d ~%d -%d (total: %d)", + watcherID, result.Added, result.Changed, result.Deleted, result.Total) + } } -// TriggerScan triggers an immediate scan for a watcher +// TriggerScan triggers an immediate differential scan for a watcher func (p *Poller) TriggerScan(watcherID string) (int, error) { w, err := p.manager.Get(watcherID) if err != nil { @@ -251,19 +255,22 @@ func (p *Poller) TriggerScan(watcherID string) (int, error) { } p.mu.Unlock() - // Perform scan - fileCount := p.scanner.ScanDirectory(w.Path) + // Perform differential scan + result := p.scanner.ScanDirectoryDiff(w.Path, watcherID) // Update status p.mu.Lock() if status, exists := p.status[watcherID]; exists { status.LastScan = NowMS() - status.FileCount = fileCount + status.FileCount = result.Total status.IsScanning = false } p.mu.Unlock() - return fileCount, nil + log.Printf("[WatcherPoller] Triggered scan for %s: +%d ~%d -%d (total: %d)", + watcherID, result.Added, result.Changed, result.Deleted, result.Total) + + return result.Total, nil } // TriggerScanAll triggers scans for all enabled watchers @@ -285,6 +292,90 @@ func (p *Poller) TriggerScanAll() (int, error) { return totalCount, nil } +// TriggerReset performs a full reset for a watcher: +// 1. Clears the Redis stream and emits a reset event +// 2. Clears the watcher's state +// 3. Performs a full scan (all files are "new") +func (p *Poller) TriggerReset(watcherID string) (int, error) { + w, err := p.manager.Get(watcherID) + if err != nil { + return 0, err + } + + if p.scanner == nil || p.dispatcher == nil { + return 0, nil + } + + log.Printf("[WatcherPoller] Triggering reset for watcher %s", watcherID) + + // Mark as scanning + p.mu.Lock() + if status, exists := p.status[watcherID]; exists { + status.IsScanning = true + } else { + p.status[watcherID] = &RuntimeStatus{IsScanning: true} + } + p.mu.Unlock() + + // Step 1: Emit reset event (this clears the stream first) + if err := p.dispatcher.EmitReset(watcherID); err != nil { + log.Printf("[WatcherPoller] Warning: failed to emit reset for %s: %v", watcherID, err) + } + + // Step 2: Clear the watcher's state + p.scanner.ClearState(watcherID) + + // Step 3: Clear event buffer + p.scanner.ClearEventBuffer() + + // Step 4: Perform full scan (all files are now "new" since state is cleared) + result := p.scanner.ScanDirectoryDiff(w.Path, watcherID) + + // Update status + p.mu.Lock() + if status, exists := p.status[watcherID]; exists { + status.LastScan = NowMS() + status.FileCount = result.Total + status.IsScanning = false + } + p.mu.Unlock() + + log.Printf("[WatcherPoller] Reset complete for %s: %d files", watcherID, result.Total) + + return result.Total, nil +} + +// TriggerResetAll performs a full reset for all enabled watchers +func (p *Poller) TriggerResetAll() (int, error) { + watchers := p.manager.List() + totalCount := 0 + isFirst := true + + for _, w := range watchers { + if w.Enabled { + // Only emit reset event once (for the first watcher) + // Subsequent watchers just clear state and rescan + if isFirst { + count, err := p.TriggerReset(w.ID) + if err != nil { + log.Printf("[WatcherPoller] Failed to reset watcher %s: %v", w.ID, err) + continue + } + totalCount += count + isFirst = false + } else { + // Clear state and rescan without clearing stream again + p.scanner.ClearState(w.ID) + result := p.scanner.ScanDirectoryDiff(w.Path, w.ID) + totalCount += result.Total + log.Printf("[WatcherPoller] Reset scan for %s: %d files", w.ID, result.Total) + } + } + } + + return totalCount, nil +} + // GetStatus returns the runtime status for a watcher func (p *Poller) GetStatus(watcherID string) *RuntimeStatus { p.mu.RLock() diff --git a/internal/webdav/handler.go b/internal/webdav/handler.go new file mode 100644 index 0000000..cde49e7 --- /dev/null +++ b/internal/webdav/handler.go @@ -0,0 +1,389 @@ +package webdav + +import ( + "encoding/json" + "io" + "log" + "net/http" + "os" + "path/filepath" + "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 +type Handler struct { + config *config.Config + cacheManager *cache.Manager + filesPath string + webdavFS webdav.Handler +} + +// NewHandler creates a new WebDAV handler with caching support +func NewHandler(cfg *config.Config, cacheManager *cache.Manager) *Handler { + h := &Handler{ + config: cfg, + cacheManager: cacheManager, + filesPath: cfg.FilesPath, + } + + // Create standard WebDAV handler for write operations + h.webdavFS = webdav.Handler{ + FileSystem: webdav.Dir(cfg.FilesPath), + LockSystem: webdav.NewMemLS(), + Logger: func(r *http.Request, err error) { + if err != nil { + log.Printf("[WebDAV] %s %s: %v", r.Method, r.URL.Path, err) + } + }, + } + + return h +} + +// ServeHTTP handles WebDAV requests +func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + // Remove /webdav prefix to get the file path + path := strings.TrimPrefix(r.URL.Path, "/webdav") + if path == "" { + path = "/" + } + + log.Printf("[WebDAV] %s %s", r.Method, path) + + switch r.Method { + case "GET", "HEAD": + h.handleRead(w, r, path) + case "PUT": + h.handlePut(w, r, path) + case "DELETE": + h.handleDelete(w, r, path) + case "MKCOL": + h.handleMkcol(w, r, path) + case "COPY", "MOVE": + h.handleCopyMove(w, r, path) + case "PROPFIND": + h.handlePropfind(w, r, path) + case "OPTIONS": + h.handleOptions(w, r) + default: + // Pass to standard WebDAV handler + h.webdavFS.ServeHTTP(w, r) + } +} + +// handleRead handles GET and HEAD requests with caching +func (h *Handler) handleRead(w http.ResponseWriter, r *http.Request, path string) { + fullPath := filepath.Join(h.filesPath, path) + + // Check if it's a directory + info, err := os.Stat(fullPath) + if err != nil { + if os.IsNotExist(err) { + http.Error(w, "Not found", http.StatusNotFound) + } else { + http.Error(w, err.Error(), http.StatusInternalServerError) + } + return + } + + if info.IsDir() { + // Return directory listing as JSON + h.handleDirListing(w, r, fullPath, path) + 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) + } + } +} + +// handleDirListing returns a directory listing as JSON +func (h *Handler) handleDirListing(w http.ResponseWriter, r *http.Request, fullPath, relPath string) { + entries, err := os.ReadDir(fullPath) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + type dirEntry struct { + Name string `json:"name"` + Type string `json:"type"` + MTime string `json:"mtime"` + Size int64 `json:"size,omitempty"` + } + + var result []dirEntry + for _, entry := range entries { + info, err := entry.Info() + if err != nil { + continue + } + + entryType := "file" + if entry.IsDir() { + entryType = "directory" + } + + result = append(result, dirEntry{ + Name: entry.Name(), + Type: entryType, + MTime: info.ModTime().Format("2006-01-02T15:04:05Z"), + Size: info.Size(), + }) + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(result) +} + +// handlePut handles PUT requests (file upload) +func (h *Handler) handlePut(w http.ResponseWriter, r *http.Request, path string) { + fullPath := filepath.Join(h.filesPath, path) + + // Ensure parent directory exists + if err := os.MkdirAll(filepath.Dir(fullPath), 0755); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + // Create/overwrite file + file, err := os.Create(fullPath) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + defer file.Close() + + // Copy request body to file + _, err = io.Copy(file, r.Body) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + // Invalidate cache for this path + h.cacheManager.Invalidate(path) + + w.WriteHeader(http.StatusCreated) +} + +// handleDelete handles DELETE requests +func (h *Handler) handleDelete(w http.ResponseWriter, r *http.Request, path string) { + fullPath := filepath.Join(h.filesPath, path) + + // Check if exists + _, err := os.Stat(fullPath) + if err != nil { + if os.IsNotExist(err) { + http.Error(w, "Not found", http.StatusNotFound) + } else { + http.Error(w, err.Error(), http.StatusInternalServerError) + } + return + } + + // Remove file or directory + if err := os.RemoveAll(fullPath); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + // Invalidate cache + h.cacheManager.Invalidate(path) + + w.WriteHeader(http.StatusNoContent) +} + +// handleMkcol handles MKCOL requests (create directory) +func (h *Handler) handleMkcol(w http.ResponseWriter, r *http.Request, path string) { + fullPath := filepath.Join(h.filesPath, path) + + // Check if already exists + if _, err := os.Stat(fullPath); err == nil { + http.Error(w, "Already exists", http.StatusMethodNotAllowed) + return + } + + // Create directory + if err := os.MkdirAll(fullPath, 0755); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + w.WriteHeader(http.StatusCreated) +} + +// handleCopyMove handles COPY and MOVE requests +func (h *Handler) handleCopyMove(w http.ResponseWriter, r *http.Request, path string) { + // Get destination from header + dest := r.Header.Get("Destination") + if dest == "" { + http.Error(w, "Destination header required", http.StatusBadRequest) + return + } + + // Parse destination URL to get path + destPath := strings.TrimPrefix(dest, "/webdav") + if strings.HasPrefix(dest, "http") { + // Full URL provided - extract path + parts := strings.SplitN(dest, "/webdav", 2) + if len(parts) == 2 { + destPath = parts[1] + } + } + + srcFullPath := filepath.Join(h.filesPath, path) + destFullPath := filepath.Join(h.filesPath, destPath) + + // Check if source exists + if _, err := os.Stat(srcFullPath); err != nil { + if os.IsNotExist(err) { + http.Error(w, "Source not found", http.StatusNotFound) + } else { + http.Error(w, err.Error(), http.StatusInternalServerError) + } + return + } + + // Ensure destination directory exists + if err := os.MkdirAll(filepath.Dir(destFullPath), 0755); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + if r.Method == "MOVE" { + // Move file + if err := os.Rename(srcFullPath, destFullPath); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + // Invalidate cache for both paths + h.cacheManager.Invalidate(path) + h.cacheManager.Invalidate(destPath) + } else { + // Copy file + if err := copyFile(srcFullPath, destFullPath); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + } + + w.WriteHeader(http.StatusCreated) +} + +// handlePropfind handles PROPFIND requests +func (h *Handler) handlePropfind(w http.ResponseWriter, r *http.Request, path string) { + fullPath := filepath.Join(h.filesPath, path) + + info, err := os.Stat(fullPath) + if err != nil { + if os.IsNotExist(err) { + http.Error(w, "Not found", http.StatusNotFound) + } else { + http.Error(w, err.Error(), http.StatusInternalServerError) + } + return + } + + depth := r.Header.Get("Depth") + if depth == "" { + depth = "infinity" + } + + // Build XML response + w.Header().Set("Content-Type", "application/xml; charset=utf-8") + w.WriteHeader(http.StatusMultiStatus) + + w.Write([]byte(`` + "\n")) + w.Write([]byte(`` + "\n")) + + // Add entry for the requested resource + h.writePropfindEntry(w, path, info) + + // If directory and depth > 0, add children + if info.IsDir() && depth != "0" { + entries, err := os.ReadDir(fullPath) + if err == nil { + for _, entry := range entries { + childInfo, err := entry.Info() + if err != nil { + continue + } + childPath := filepath.Join(path, entry.Name()) + h.writePropfindEntry(w, childPath, childInfo) + } + } + } + + w.Write([]byte(`` + "\n")) +} + +// writePropfindEntry writes a single PROPFIND response entry +func (h *Handler) writePropfindEntry(w http.ResponseWriter, path string, info os.FileInfo) { + href := "/webdav" + path + if info.IsDir() && !strings.HasSuffix(href, "/") { + href += "/" + } + + resourceType := "" + if info.IsDir() { + resourceType = "" + } + + w.Write([]byte(` ` + "\n")) + w.Write([]byte(` ` + href + `` + "\n")) + w.Write([]byte(` ` + "\n")) + w.Write([]byte(` ` + "\n")) + w.Write([]byte(` ` + resourceType + `` + "\n")) + if !info.IsDir() { + w.Write([]byte(` ` + formatInt64(info.Size()) + `` + "\n")) + } + w.Write([]byte(` ` + info.ModTime().Format("Mon, 02 Jan 2006 15:04:05 GMT") + `` + "\n")) + w.Write([]byte(` ` + "\n")) + w.Write([]byte(` HTTP/1.1 200 OK` + "\n")) + w.Write([]byte(` ` + "\n")) + w.Write([]byte(` ` + "\n")) +} + +// handleOptions handles OPTIONS requests +func (h *Handler) handleOptions(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Allow", "OPTIONS, GET, HEAD, PUT, DELETE, MKCOL, COPY, MOVE, PROPFIND") + w.Header().Set("DAV", "1, 2") + w.Header().Set("MS-Author-Via", "DAV") + w.WriteHeader(http.StatusOK) +} + +// copyFile copies a file from src to dst +func copyFile(src, dst string) error { + srcFile, err := os.Open(src) + if err != nil { + return err + } + defer srcFile.Close() + + dstFile, err := os.Create(dst) + if err != nil { + return err + } + defer dstFile.Close() + + _, err = io.Copy(dstFile, srcFile) + return err +} + +// formatInt64 formats an int64 as a string +func formatInt64(n int64) string { + return strconv.FormatInt(n, 10) +}