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

code cleanup

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