metastar/meta-core
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

new flat KV structure

Browse Source
This commit is contained in:
Pierre
2026-02-08 21:53:42 +01:00
parent 4c3a0df4b4
commit 79d86b62d3
20 changed files with 2335 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docker/nginx.conf
+22 -23
View File
@@ -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;
}
}
}
go.mod
+1 -1
View File
@@ -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
)
go.sum
+2 -2
View File
@@ -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=
internal/api/server.go
+78
View File
@@ -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
}
internal/cache/handlers.go
Vendored
+73
View File
@@ -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,
})
}
internal/cache/index.go
Vendored
+242
View File
@@ -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
}
internal/cache/invalidator.go
Vendored
+166
View File
@@ -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)
}
}
internal/cache/manager.go
Vendored
+494
View File
@@ -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
}
internal/cache/types.go
Vendored
+48
View File
@@ -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(),
}
}
internal/config/config.go
+22
View File
@@ -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"
internal/events/meta_publisher.go
+139
View File
@@ -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)
}
}
internal/leader/redis.go
+5
View File
@@ -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...)
internal/storage/client.go
+165 -65
View File
@@ -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
}
internal/watcher/dispatcher.go
+49
View File
@@ -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)
}
internal/watcher/state.go
+146
View File
@@ -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
}
internal/watcher/types.go
+10
View File
@@ -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()
internal/watcher/watcher.go
+143 -9
View File
@@ -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) {
internal/watchers/handlers.go
+40
View File
@@ -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{}) {
internal/watchers/poller.go
+101 -10
View File
@@ -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()
internal/webdav/handler.go
+389
View File
@@ -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(`<?xml version="1.0" encoding="utf-8"?>` + "\n"))
w.Write([]byte(`<D:multistatus xmlns:D="DAV:">` + "\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(`</D:multistatus>` + "\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 = "<D:collection/>"
}
w.Write([]byte(` <D:response>` + "\n"))
w.Write([]byte(` <D:href>` + href + `</D:href>` + "\n"))
w.Write([]byte(` <D:propstat>` + "\n"))
w.Write([]byte(` <D:prop>` + "\n"))
w.Write([]byte(` <D:resourcetype>` + resourceType + `</D:resourcetype>` + "\n"))
if !info.IsDir() {
w.Write([]byte(` <D:getcontentlength>` + formatInt64(info.Size()) + `</D:getcontentlength>` + "\n"))
}
w.Write([]byte(` <D:getlastmodified>` + info.ModTime().Format("Mon, 02 Jan 2006 15:04:05 GMT") + `</D:getlastmodified>` + "\n"))
w.Write([]byte(` </D:prop>` + "\n"))
w.Write([]byte(` <D:status>HTTP/1.1 200 OK</D:status>` + "\n"))
w.Write([]byte(` </D:propstat>` + "\n"))
w.Write([]byte(` </D:response>` + "\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)
}