diff --git a/dashboard/src/App.tsx b/dashboard/src/App.tsx index ece0bfa..dce7294 100644 --- a/dashboard/src/App.tsx +++ b/dashboard/src/App.tsx @@ -1,6 +1,7 @@ import { BrowserRouter, Routes, Route, NavLink } from 'react-router-dom'; import Dashboard from './pages/Dashboard'; import Mounts from './pages/Mounts'; +import Watchers from './pages/Watchers'; import Files from './pages/Files'; import ServiceNav from './components/ServiceNav'; @@ -61,6 +62,12 @@ export default function App() { > Mounts + (isActive ? activeLinkStyle : linkStyle)} + > + Watchers + (isActive ? activeLinkStyle : linkStyle)} @@ -80,6 +87,7 @@ export default function App() { } /> } /> + } /> } /> } /> diff --git a/dashboard/src/pages/Dashboard.tsx b/dashboard/src/pages/Dashboard.tsx index 59145b9..c76bbc0 100644 --- a/dashboard/src/pages/Dashboard.tsx +++ b/dashboard/src/pages/Dashboard.tsx @@ -7,11 +7,20 @@ interface HealthStatus { timestamp: string; } -interface ScanStatus { - status: string; - scanning: boolean; +interface WatcherStatus { + id: string; + path: string; + intervalMs: number; + enabled: boolean; + active: boolean; lastScan: number; fileCount: number; + isScanning: boolean; +} + +interface WatchersResponse { + watchers: WatcherStatus[]; + count: number; } interface ServiceInfo { @@ -42,7 +51,7 @@ const statStyle: React.CSSProperties = { export default function Dashboard() { const [health, setHealth] = useState(null); - const [scanStatus, setScanStatus] = useState(null); + const [watchersData, setWatchersData] = useState(null); const [services, setServices] = useState([]); const [error, setError] = useState(null); @@ -55,10 +64,10 @@ export default function Dashboard() { setHealth(await healthRes.json()); } - // Fetch scan status - const scanRes = await fetch('/api/scan/status'); - if (scanRes.ok) { - setScanStatus(await scanRes.json()); + // Fetch watchers status + const watchersRes = await fetch('/api/watchers'); + if (watchersRes.ok) { + setWatchersData(await watchersRes.json()); } // Fetch services @@ -81,12 +90,18 @@ export default function Dashboard() { const handleTriggerScan = async () => { try { - await fetch('/api/scan/trigger', { method: 'POST' }); + await fetch('/api/watchers/scan-all', { method: 'POST' }); } catch (err) { console.error('Failed to trigger scan:', err); } }; + // Aggregate watcher stats + const totalFiles = watchersData?.watchers.reduce((sum, w) => sum + w.fileCount, 0) || 0; + const isScanning = watchersData?.watchers.some(w => w.isScanning) || false; + const activeWatchers = watchersData?.watchers.filter(w => w.active).length || 0; + const lastScan = watchersData?.watchers.reduce((max, w) => Math.max(max, w.lastScan || 0), 0) || 0; + return (

meta-core Dashboard

@@ -114,16 +129,17 @@ export default function Dashboard() { )}
- {/* Scan Status Card */} + {/* Watchers Status Card */}
-

File Watcher

- {scanStatus ? ( +

File Watchers

+ {watchersData ? ( <> -
{scanStatus.fileCount}
+
{totalFiles}

Files discovered

-

Status: {scanStatus.scanning ? 'Scanning...' : scanStatus.status}

- {scanStatus.lastScan > 0 && ( -

Last scan: {new Date(scanStatus.lastScan).toLocaleString()}

+

Watchers: {activeWatchers} active / {watchersData.count} total

+

Status: {isScanning ? 'Scanning...' : 'Idle'}

+ {lastScan > 0 && ( +

Last scan: {new Date(lastScan).toLocaleString()}

)} ) : ( diff --git a/dashboard/src/pages/Watchers.tsx b/dashboard/src/pages/Watchers.tsx new file mode 100644 index 0000000..ce59aa3 --- /dev/null +++ b/dashboard/src/pages/Watchers.tsx @@ -0,0 +1,353 @@ +import { useEffect, useState } from 'react'; + +interface Watcher { + id: string; + path: string; + intervalMs: number; + enabled: boolean; + createdAt: number; + active: boolean; + lastScan: number; + fileCount: number; + isScanning: boolean; +} + +const cardStyle: React.CSSProperties = { + background: '#16213e', + borderRadius: '8px', + padding: '1.5rem', + marginBottom: '1rem', +}; + +const buttonStyle: React.CSSProperties = { + padding: '0.5rem 1rem', + background: '#0f3460', + color: '#fff', + border: 'none', + borderRadius: '4px', + cursor: 'pointer', + marginRight: '0.5rem', +}; + +const dangerButtonStyle: React.CSSProperties = { + ...buttonStyle, + background: '#7f1d1d', +}; + +const inputStyle: React.CSSProperties = { + width: '100%', + padding: '0.5rem', + marginBottom: '0.5rem', + background: '#1a1a2e', + border: '1px solid #0f3460', + borderRadius: '4px', + color: '#fff', +}; + +function formatTimeAgo(timestamp: number): string { + if (!timestamp) return 'Never'; + const seconds = Math.floor((Date.now() - timestamp) / 1000); + if (seconds < 60) return `${seconds}s ago`; + const minutes = Math.floor(seconds / 60); + if (minutes < 60) return `${minutes}m ago`; + const hours = Math.floor(minutes / 60); + if (hours < 24) return `${hours}h ago`; + return new Date(timestamp).toLocaleDateString(); +} + +export default function Watchers() { + const [watchers, setWatchers] = useState([]); + const [loading, setLoading] = useState(true); + const [showForm, setShowForm] = useState(false); + const [editingId, setEditingId] = useState(null); + const [formData, setFormData] = useState({ + path: '', + intervalMs: 30000, + enabled: true, + }); + const [scanning, setScanning] = useState(null); + const [scanningAll, setScanningAll] = useState(false); + + const fetchWatchers = async () => { + try { + const res = await fetch('/api/watchers'); + if (res.ok) { + const data = await res.json(); + setWatchers(data.watchers || []); + } + } catch (err) { + console.error('Failed to fetch watchers:', err); + } finally { + setLoading(false); + } + }; + + useEffect(() => { + fetchWatchers(); + const interval = setInterval(fetchWatchers, 5000); + return () => clearInterval(interval); + }, []); + + const handleCreate = async (e: React.FormEvent) => { + e.preventDefault(); + + const body = { + path: formData.path, + intervalMs: formData.intervalMs, + enabled: formData.enabled, + }; + + const res = await fetch('/api/watchers', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify(body), + }); + + if (res.ok) { + setShowForm(false); + setFormData({ path: '', intervalMs: 30000, enabled: true }); + fetchWatchers(); + } + }; + + const handleUpdate = async (e: React.FormEvent) => { + e.preventDefault(); + if (!editingId) return; + + const body = { + path: formData.path, + intervalMs: formData.intervalMs, + enabled: formData.enabled, + }; + + const res = await fetch(`/api/watchers/${editingId}`, { + method: 'PUT', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify(body), + }); + + if (res.ok) { + setEditingId(null); + setFormData({ path: '', intervalMs: 30000, enabled: true }); + fetchWatchers(); + } + }; + + const handleDelete = async (id: string) => { + if (!confirm('Are you sure you want to delete this watcher?')) return; + await fetch(`/api/watchers/${id}`, { method: 'DELETE' }); + fetchWatchers(); + }; + + const handleScan = async (id: string) => { + setScanning(id); + try { + await fetch(`/api/watchers/${id}/scan`, { method: 'POST' }); + fetchWatchers(); + } finally { + setScanning(null); + } + }; + + const handleScanAll = async () => { + setScanningAll(true); + try { + await fetch('/api/watchers/scan-all', { method: 'POST' }); + fetchWatchers(); + } finally { + setScanningAll(false); + } + }; + + const handleToggleEnabled = async (id: string, enabled: boolean) => { + await fetch(`/api/watchers/${id}`, { + method: 'PUT', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ enabled }), + }); + fetchWatchers(); + }; + + const startEdit = (watcher: Watcher) => { + setEditingId(watcher.id); + setFormData({ + path: watcher.path, + intervalMs: watcher.intervalMs, + enabled: watcher.enabled, + }); + setShowForm(false); + }; + + const cancelEdit = () => { + setEditingId(null); + setFormData({ path: '', intervalMs: 30000, enabled: true }); + }; + + if (loading) return
Loading...
; + + return ( +
+
+

Watchers

+
+ + +
+
+ + {showForm && ( +
+

Add New Watcher

+
+ setFormData({ ...formData, path: e.target.value })} + required + /> +
+ setFormData({ ...formData, intervalMs: parseInt(e.target.value) * 1000 || 30000 })} + /> + seconds +
+ + +
+
+ )} + + {watchers.length === 0 ? ( +
+

No watchers configured. Click "Add Watcher" to get started.

+
+ ) : ( + watchers.map((watcher) => ( +
+ {editingId === watcher.id ? ( +
+

Edit Watcher

+ setFormData({ ...formData, path: e.target.value })} + required + /> +
+ setFormData({ ...formData, intervalMs: parseInt(e.target.value) * 1000 || 30000 })} + /> + seconds +
+ + + +
+ ) : ( +
+
+

{watcher.path}

+

+ Interval: {watcher.intervalMs / 1000}s + {' | '} + Last scan: {formatTimeAgo(watcher.lastScan)} + {watcher.fileCount > 0 && ` | ${watcher.fileCount} files`} +

+

+ Status:{' '} + {watcher.isScanning ? ( + Scanning... + ) : watcher.active ? ( + Active + ) : watcher.enabled ? ( + Pending + ) : ( + Disabled + )} +

+
+
+
+ +
+
+ + + +
+
+
+ )} +
+ )) + )} +
+ ); +} diff --git a/go.mod b/go.mod index ebde4cd..a06e419 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,6 @@ module github.com/metazla/meta-core go 1.21 require ( - github.com/fsnotify/fsnotify v1.7.0 github.com/google/uuid v1.6.0 github.com/gorilla/mux v1.8.1 github.com/redis/go-redis/v9 v9.4.0 diff --git a/go.sum b/go.sum index 130292a..be30923 100644 --- a/go.sum +++ b/go.sum @@ -6,8 +6,6 @@ github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78= github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc= -github.com/fsnotify/fsnotify v1.7.0 h1:8JEhPFa5W2WU7YfeZzPNqzMP6Lwt7L2715Ggo0nosvA= -github.com/fsnotify/fsnotify v1.7.0/go.mod h1:40Bi/Hjc2AVfZrqy+aj+yEI+/bRxZnMJyTJwOpGvigM= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/gorilla/mux v1.8.1 h1:TuBL49tXwgrFYWhqrNgrUNEY92u81SPhu7sTdzQEiWY= diff --git a/internal/api/server.go b/internal/api/server.go index 42722fb..a26b46e 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -14,6 +14,7 @@ import ( "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" ) // Server is the HTTP API server for meta-core @@ -29,6 +30,9 @@ type Server struct { watcherDispatcher *watcher.Dispatcher fileWatcher *watcher.Watcher watcherHandlers *watcher.Handlers + watchersManager *watchers.Manager + watchersPoller *watchers.Poller + watchersHandlers *watchers.Handlers router *mux.Router server *http.Server } @@ -50,19 +54,37 @@ func NewServer( router: mux.NewRouter(), } - // Initialize file watcher (if enabled) - before mounts so we can pass scanFunc + // Initialize file watcher/scanner and dispatcher var scanFunc mounts.ScanFunc - if cfg.EnableFileWatcher && len(cfg.WatchFolderList) > 0 { + if cfg.EnableFileWatcher { s.watcherDispatcher = watcher.NewDispatcher() + // Wire storage client for Redis Stream publishing + if stor != nil { + s.watcherDispatcher.SetStorageClient(stor) + } fileWatcher, err := watcher.NewWatcher(cfg, s.watcherDispatcher) if err != nil { log.Printf("[API] Warning: failed to initialize file watcher: %v", err) } else { s.fileWatcher = fileWatcher s.watcherHandlers = watcher.NewHandlers(fileWatcher, s.watcherDispatcher) - // Create scan function from watcher + // Create scan function from watcher for mounts scanFunc = fileWatcher.ScanMountPath } + + // Initialize watchers manager (polling-based file watching) + watchersManager, err := watchers.NewManager(cfg) + if err != nil { + log.Printf("[API] Warning: failed to initialize watchers manager: %v", err) + } else { + s.watchersManager = watchersManager + + // Create watchers poller + if s.fileWatcher != nil { + s.watchersPoller = watchers.NewPoller(watchersManager, s.fileWatcher, s.watcherDispatcher) + s.watchersHandlers = watchers.NewHandlers(watchersManager, s.watchersPoller) + } + } } // Initialize mounts manager @@ -151,6 +173,12 @@ func (s *Server) setupRoutes() { log.Println("[API] File watcher routes registered at /api/events/*, /api/scan/*") } + // Watchers management routes (if watchers initialized) + if s.watchersHandlers != nil { + s.watchersHandlers.RegisterRoutes(s.router) + log.Println("[API] Watchers management routes registered at /api/watchers/*") + } + // Add middleware s.router.Use(loggingMiddleware) s.router.Use(corsMiddleware) @@ -176,10 +204,10 @@ func (s *Server) Start() error { } }() - // Start file watcher (if initialized) - if s.fileWatcher != nil { - if err := s.fileWatcher.Start(); err != nil { - log.Printf("[API] Warning: failed to start file watcher: %v", err) + // Start watchers poller (if initialized) + if s.watchersPoller != nil { + if err := s.watchersPoller.Start(); err != nil { + log.Printf("[API] Warning: failed to start watchers poller: %v", err) } } @@ -195,6 +223,13 @@ func (s *Server) Start() error { // Stop gracefully stops the HTTP server func (s *Server) Stop() error { + // Stop watchers poller + if s.watchersPoller != nil { + if err := s.watchersPoller.Stop(); err != nil { + log.Printf("[API] Warning: failed to stop watchers poller: %v", err) + } + } + // Stop mount poller if s.mountPoller != nil { if err := s.mountPoller.Stop(); err != nil { @@ -202,13 +237,6 @@ func (s *Server) Stop() error { } } - // Stop file watcher - if s.fileWatcher != nil { - if err := s.fileWatcher.Stop(); err != nil { - log.Printf("[API] Warning: failed to stop file watcher: %v", err) - } - } - if s.server == nil { return nil } diff --git a/internal/config/config.go b/internal/config/config.go index c3081a5..38c3338 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -104,6 +104,11 @@ func (c *Config) MountsErrorDir() string { return c.MountsDir + "/errors" } +// WatchersFilePath returns the path to the watchers configuration file +func (c *Config) WatchersFilePath() string { + return c.MetaCorePath + "/watchers.json" +} + func getEnv(key, defaultValue string) string { if value := os.Getenv(key); value != "" { return value diff --git a/internal/storage/client.go b/internal/storage/client.go index e0355a8..5ae8ec4 100644 --- a/internal/storage/client.go +++ b/internal/storage/client.go @@ -570,3 +570,32 @@ func (c *Client) GetRedisClient() *redis.Client { func (c *Client) GetPrefix() string { return c.prefix } + +// XAdd publishes an event to a Redis Stream +// Uses XADD with MAXLEN ~ for approximate trimming +func (c *Client) XAdd(stream string, maxLen int64, fields map[string]interface{}) (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() + + // Build XADD arguments with approximate MAXLEN + args := &redis.XAddArgs{ + Stream: stream, + MaxLen: maxLen, + Approx: true, + Values: fields, + } + + id, err := c.client.XAdd(ctx, args).Result() + if err != nil { + return "", fmt.Errorf("xadd failed: %w", err) + } + + return id, nil +} diff --git a/internal/watcher/dispatcher.go b/internal/watcher/dispatcher.go index fd88880..761e4f0 100644 --- a/internal/watcher/dispatcher.go +++ b/internal/watcher/dispatcher.go @@ -1,223 +1,75 @@ package watcher import ( - "bytes" - "encoding/json" + "fmt" "log" - "net/http" "sync" - "time" + + "github.com/metazla/meta-core/internal/storage" ) const ( - // MaxRetries is the maximum number of delivery retries - MaxRetries = 3 - // RetryDelay is the delay between retries - RetryDelay = 5 * time.Second - // WebhookTimeout is the timeout for webhook requests - WebhookTimeout = 10 * time.Second - // MaxFailCount is the max failures before removing subscriber - MaxFailCount = 10 + // EventsStream is the Redis Stream name for file events + EventsStream = "file:events" + // StreamMaxLen is the approximate max length for the stream + StreamMaxLen = 10000 ) -// Dispatcher sends file events to subscribers +// Dispatcher sends file events to Redis Stream type Dispatcher struct { - subscribers map[string]*Subscriber - sseClients map[chan FileEvent]bool - mu sync.RWMutex - httpClient *http.Client + mu sync.RWMutex + storageClient *storage.Client } // NewDispatcher creates a new event dispatcher func NewDispatcher() *Dispatcher { - return &Dispatcher{ - subscribers: make(map[string]*Subscriber), - sseClients: make(map[chan FileEvent]bool), - httpClient: &http.Client{ - Timeout: WebhookTimeout, - }, - } + return &Dispatcher{} } -// Subscribe registers a webhook subscriber -func (d *Dispatcher) Subscribe(url string, eventTypes []string) error { +// SetStorageClient sets the storage client for Redis Stream publishing +func (d *Dispatcher) SetStorageClient(client *storage.Client) { d.mu.Lock() defer d.mu.Unlock() - - d.subscribers[url] = &Subscriber{ - URL: url, - RegisteredAt: NowMS(), - EventTypes: eventTypes, - FailCount: 0, - } - - log.Printf("[Dispatcher] Subscribed webhook: %s", url) - return nil + d.storageClient = client + log.Println("[Dispatcher] Storage client set for Redis Stream publishing") } -// Unsubscribe removes a webhook subscriber -func (d *Dispatcher) Unsubscribe(url string) error { - d.mu.Lock() - defer d.mu.Unlock() - - delete(d.subscribers, url) - log.Printf("[Dispatcher] Unsubscribed webhook: %s", url) - return nil -} - -// ListSubscribers returns all subscribers -func (d *Dispatcher) ListSubscribers() []Subscriber { - d.mu.RLock() - defer d.mu.RUnlock() - - result := make([]Subscriber, 0, len(d.subscribers)) - for _, sub := range d.subscribers { - result = append(result, *sub) - } - return result -} - -// AddSSEClient adds an SSE client channel -func (d *Dispatcher) AddSSEClient(ch chan FileEvent) { - d.mu.Lock() - defer d.mu.Unlock() - d.sseClients[ch] = true -} - -// RemoveSSEClient removes an SSE client channel -func (d *Dispatcher) RemoveSSEClient(ch chan FileEvent) { - d.mu.Lock() - defer d.mu.Unlock() - delete(d.sseClients, ch) - close(ch) -} - -// Dispatch sends an event to all subscribers +// Dispatch sends an event to Redis Stream func (d *Dispatcher) Dispatch(event FileEvent) { - // Dispatch to webhooks - go d.dispatchToWebhooks(event) - - // Dispatch to SSE clients - d.dispatchToSSE(event) + go d.dispatchToStream(event) } -// dispatchToWebhooks sends event to webhook subscribers -func (d *Dispatcher) dispatchToWebhooks(event FileEvent) { +// dispatchToStream publishes event to Redis Stream for service consumption +func (d *Dispatcher) dispatchToStream(event FileEvent) { d.mu.RLock() - subscribers := make([]*Subscriber, 0, len(d.subscribers)) - for _, sub := range d.subscribers { - subscribers = append(subscribers, sub) - } + client := d.storageClient d.mu.RUnlock() - for _, sub := range subscribers { - // Check event type filter - if len(sub.EventTypes) > 0 { - found := false - for _, et := range sub.EventTypes { - if et == string(event.Type) { - found = true - break - } - } - if !found { - continue - } - } - - go d.deliverToWebhook(sub.URL, event) - } -} - -// deliverToWebhook delivers an event to a webhook with retries -func (d *Dispatcher) deliverToWebhook(url string, event FileEvent) { - body, err := json.Marshal(event) - if err != nil { - log.Printf("[Dispatcher] Failed to marshal event: %v", err) + if client == nil || !client.IsConnected() { return } - var lastErr error - for attempt := 0; attempt < MaxRetries; attempt++ { - if attempt > 0 { - time.Sleep(RetryDelay) - } - - req, err := http.NewRequest("POST", url, bytes.NewReader(body)) - if err != nil { - lastErr = err - continue - } - - req.Header.Set("Content-Type", "application/json") - req.Header.Set("X-Event-Type", string(event.Type)) - - resp, err := d.httpClient.Do(req) - if err != nil { - lastErr = err - continue - } - - resp.Body.Close() - - if resp.StatusCode >= 200 && resp.StatusCode < 300 { - // Success - d.mu.Lock() - if sub, ok := d.subscribers[url]; ok { - sub.LastDelivery = NowMS() - sub.FailCount = 0 - } - d.mu.Unlock() - return - } - - lastErr = &httpError{StatusCode: resp.StatusCode} + // Build stream fields + fields := map[string]interface{}{ + "type": string(event.Type), + "path": event.Path, + "timestamp": fmt.Sprintf("%d", event.Timestamp), } - // All retries failed - log.Printf("[Dispatcher] Failed to deliver to %s after %d retries: %v", url, MaxRetries, lastErr) - - d.mu.Lock() - if sub, ok := d.subscribers[url]; ok { - sub.FailCount++ - if sub.FailCount >= MaxFailCount { - delete(d.subscribers, url) - log.Printf("[Dispatcher] Removed webhook %s after %d failures", url, MaxFailCount) - } + // Add optional fields + if event.Size > 0 { + fields["size"] = fmt.Sprintf("%d", event.Size) } - d.mu.Unlock() -} - -// dispatchToSSE sends event to SSE clients -func (d *Dispatcher) dispatchToSSE(event FileEvent) { - d.mu.RLock() - clients := make([]chan FileEvent, 0, len(d.sseClients)) - for ch := range d.sseClients { - clients = append(clients, ch) + if event.PartialHash != "" { + fields["partialHash"] = event.PartialHash + } + if event.OldPath != "" { + fields["oldPath"] = event.OldPath } - d.mu.RUnlock() - for _, ch := range clients { - select { - case ch <- event: - default: - // Channel full, skip - } + // Publish to stream + _, err := client.XAdd(EventsStream, StreamMaxLen, fields) + if err != nil { + log.Printf("[Dispatcher] Failed to publish to stream: %v", err) } } - -// SSEClientCount returns the number of connected SSE clients -func (d *Dispatcher) SSEClientCount() int { - d.mu.RLock() - defer d.mu.RUnlock() - return len(d.sseClients) -} - -// httpError represents an HTTP error -type httpError struct { - StatusCode int -} - -func (e *httpError) Error() string { - return http.StatusText(e.StatusCode) -} diff --git a/internal/watcher/handlers.go b/internal/watcher/handlers.go index ecb6664..ef2ce11 100644 --- a/internal/watcher/handlers.go +++ b/internal/watcher/handlers.go @@ -2,7 +2,6 @@ package watcher import ( "encoding/json" - "fmt" "net/http" "strconv" @@ -25,59 +24,12 @@ func NewHandlers(watcher *Watcher, dispatcher *Dispatcher) *Handlers { // RegisterRoutes registers all watcher-related routes func (h *Handlers) RegisterRoutes(r *mux.Router) { - // Event subscriptions - r.HandleFunc("/api/events/subscribe", h.handleSSESubscribe).Methods("GET") + // Event polling (for services that don't use Redis Streams) r.HandleFunc("/api/events/poll", h.handlePoll).Methods("GET") - r.HandleFunc("/api/events/subscribers", h.handleListSubscribers).Methods("GET") - r.HandleFunc("/api/events/subscribers", h.handleAddSubscriber).Methods("POST") - r.HandleFunc("/api/events/subscribers/{url}", h.handleRemoveSubscriber).Methods("DELETE") - // Scan management - r.HandleFunc("/api/scan/trigger", h.handleTriggerScan).Methods("POST") - r.HandleFunc("/api/scan/status", h.handleScanStatus).Methods("GET") -} - -// handleSSESubscribe handles GET /api/events/subscribe (Server-Sent Events) -func (h *Handlers) handleSSESubscribe(w http.ResponseWriter, r *http.Request) { - // Set SSE headers - w.Header().Set("Content-Type", "text/event-stream") - w.Header().Set("Cache-Control", "no-cache") - w.Header().Set("Connection", "keep-alive") - w.Header().Set("Access-Control-Allow-Origin", "*") - - // Create event channel - eventChan := make(chan FileEvent, 100) - h.dispatcher.AddSSEClient(eventChan) - - // Cleanup on disconnect - defer h.dispatcher.RemoveSSEClient(eventChan) - - // Get flusher - flusher, ok := w.(http.Flusher) - if !ok { - http.Error(w, "SSE not supported", http.StatusInternalServerError) - return - } - - // Send initial connected message - fmt.Fprintf(w, "event: connected\ndata: {\"status\":\"connected\"}\n\n") - flusher.Flush() - - // Stream events - for { - select { - case event := <-eventChan: - data, err := json.Marshal(event) - if err != nil { - continue - } - fmt.Fprintf(w, "event: file\ndata: %s\n\n", string(data)) - flusher.Flush() - - case <-r.Context().Done(): - return - } - } + // Deprecated scan endpoints - redirect to /api/watchers + r.HandleFunc("/api/scan/trigger", h.handleTriggerScanDeprecated).Methods("POST") + r.HandleFunc("/api/scan/status", h.handleScanStatusDeprecated).Methods("GET") } // handlePoll handles GET /api/events/poll @@ -108,78 +60,22 @@ func (h *Handlers) handlePoll(w http.ResponseWriter, r *http.Request) { }) } -// handleListSubscribers handles GET /api/events/subscribers -func (h *Handlers) handleListSubscribers(w http.ResponseWriter, r *http.Request) { - subscribers := h.dispatcher.ListSubscribers() - - writeJSON(w, http.StatusOK, SubscribersListResponse{ - Subscribers: subscribers, - Count: len(subscribers), - }) -} - -// handleAddSubscriber handles POST /api/events/subscribers -func (h *Handlers) handleAddSubscriber(w http.ResponseWriter, r *http.Request) { - var req SubscribeRequest - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - writeError(w, http.StatusBadRequest, "invalid JSON body") - return - } - - if req.URL == "" { - writeError(w, http.StatusBadRequest, "URL is required") - return - } - - if err := h.dispatcher.Subscribe(req.URL, req.EventTypes); err != nil { - writeError(w, http.StatusInternalServerError, err.Error()) - return - } - - writeJSON(w, http.StatusCreated, map[string]interface{}{ - "status": "ok", - "message": "Subscribed", - "url": req.URL, - }) -} - -// handleRemoveSubscriber handles DELETE /api/events/subscribers/{url} -func (h *Handlers) handleRemoveSubscriber(w http.ResponseWriter, r *http.Request) { - vars := mux.Vars(r) - url := vars["url"] - - if url == "" { - writeError(w, http.StatusBadRequest, "URL is required") - return - } - - if err := h.dispatcher.Unsubscribe(url); err != nil { - writeError(w, http.StatusInternalServerError, err.Error()) - return - } - +// handleTriggerScanDeprecated handles POST /api/scan/trigger (deprecated) +func (h *Handlers) handleTriggerScanDeprecated(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, map[string]interface{}{ "status": "ok", - "message": "Unsubscribed", + "message": "Use POST /api/watchers/scan-all instead. This endpoint is deprecated.", }) } -// handleTriggerScan handles POST /api/scan/trigger -func (h *Handlers) handleTriggerScan(w http.ResponseWriter, r *http.Request) { - go h.watcher.RunScan() - +// handleScanStatusDeprecated handles GET /api/scan/status (deprecated) +func (h *Handlers) handleScanStatusDeprecated(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, map[string]interface{}{ - "status": "ok", - "message": "Scan triggered", + "status": "deprecated", + "message": "Use GET /api/watchers instead. This endpoint is deprecated.", }) } -// handleScanStatus handles GET /api/scan/status -func (h *Handlers) handleScanStatus(w http.ResponseWriter, r *http.Request) { - status := h.watcher.GetStatus() - writeJSON(w, http.StatusOK, status) -} - // Helper functions func writeJSON(w http.ResponseWriter, status int, data interface{}) { diff --git a/internal/watcher/types.go b/internal/watcher/types.go index c399c21..4a211d5 100644 --- a/internal/watcher/types.go +++ b/internal/watcher/types.go @@ -22,15 +22,6 @@ type FileEvent struct { OldPath string `json:"oldPath,omitempty"` // For rename events } -// Subscriber represents a webhook subscriber -type Subscriber struct { - URL string `json:"url"` - RegisteredAt int64 `json:"registeredAt"` - EventTypes []string `json:"eventTypes,omitempty"` // Empty means all - LastDelivery int64 `json:"lastDelivery,omitempty"` - FailCount int `json:"failCount"` -} - // PendingEvent tracks a file event that's being debounced type PendingEvent struct { Event FileEvent @@ -39,24 +30,12 @@ type PendingEvent struct { Timer *time.Timer } -// SubscribeRequest is the request to register a webhook -type SubscribeRequest struct { - URL string `json:"url"` - EventTypes []string `json:"eventTypes,omitempty"` -} - // EventsListResponse is the response for listing events type EventsListResponse struct { Events []FileEvent `json:"events"` Count int `json:"count"` } -// SubscribersListResponse is the response for listing subscribers -type SubscribersListResponse struct { - Subscribers []Subscriber `json:"subscribers"` - Count int `json:"count"` -} - // ScanStatusResponse is the response for scan status type ScanStatusResponse struct { Status string `json:"status"` diff --git a/internal/watcher/watcher.go b/internal/watcher/watcher.go index 1bdb3b2..b5620a0 100644 --- a/internal/watcher/watcher.go +++ b/internal/watcher/watcher.go @@ -3,16 +3,13 @@ package watcher import ( "crypto/sha256" "encoding/hex" - "fmt" "io" "log" "os" "path/filepath" "strings" "sync" - "time" - "github.com/fsnotify/fsnotify" "github.com/metazla/meta-core/internal/config" ) @@ -21,188 +18,32 @@ const ( PartialHashSize = 64 * 1024 // 64KB ) -// Watcher monitors directories for file changes +// Watcher provides file scanning and event dispatching type Watcher struct { config *config.Config - fsWatcher *fsnotify.Watcher debouncer *Debouncer dispatcher *Dispatcher filesPath string - watchPaths []string mu sync.RWMutex - isRunning bool - isScanning bool - lastScan int64 - fileCount int - stopChan chan struct{} eventBuffer []FileEvent } -// NewWatcher creates a new file watcher +// NewWatcher creates a new file watcher/scanner func NewWatcher(cfg *config.Config, dispatcher *Dispatcher) (*Watcher, error) { - fsWatcher, err := fsnotify.NewWatcher() - if err != nil { - return nil, err - } - - debouncer := NewDebouncer(time.Duration(cfg.DebounceMS) * time.Millisecond) + debouncer := NewDebouncer(0) // No debouncing for polling-based scanning w := &Watcher{ config: cfg, - fsWatcher: fsWatcher, debouncer: debouncer, dispatcher: dispatcher, filesPath: cfg.FilesPath, - watchPaths: cfg.WatchFolderList, - stopChan: make(chan struct{}), eventBuffer: make([]FileEvent, 0, 1000), } - // Set debouncer callback - debouncer.SetCallback(func(event FileEvent) { - w.handleDebouncedEvent(event) - }) - return w, nil } -// Start begins watching for file changes -func (w *Watcher) Start() error { - w.mu.Lock() - if w.isRunning { - w.mu.Unlock() - return nil - } - w.isRunning = true - w.mu.Unlock() - - // Add watch paths - for _, watchPath := range w.watchPaths { - if err := w.addWatchRecursive(watchPath); err != nil { - log.Printf("[Watcher] Warning: failed to watch %s: %v", watchPath, err) - } - } - - // Start event processing goroutine - go w.processEvents() - - // Start initial scan - go w.RunScan() - - log.Printf("[Watcher] Started watching %d paths", len(w.watchPaths)) - return nil -} - -// Stop stops the watcher -func (w *Watcher) Stop() error { - w.mu.Lock() - if !w.isRunning { - w.mu.Unlock() - return nil - } - w.isRunning = false - w.mu.Unlock() - - close(w.stopChan) - w.debouncer.Stop() - return w.fsWatcher.Close() -} - -// addWatchRecursive adds a directory and all subdirectories to the watch list -func (w *Watcher) addWatchRecursive(root string) error { - return filepath.Walk(root, func(path string, info os.FileInfo, err error) error { - if err != nil { - return nil // Skip inaccessible paths - } - if info.IsDir() { - if err := w.fsWatcher.Add(path); err != nil { - log.Printf("[Watcher] Warning: cannot watch %s: %v", path, err) - } - } - return nil - }) -} - -// processEvents handles fsnotify events -func (w *Watcher) processEvents() { - for { - select { - case <-w.stopChan: - return - - case event, ok := <-w.fsWatcher.Events: - if !ok { - return - } - w.handleFsEvent(event) - - case err, ok := <-w.fsWatcher.Errors: - if !ok { - return - } - log.Printf("[Watcher] Error: %v", err) - } - } -} - -// handleFsEvent converts an fsnotify event to a FileEvent -func (w *Watcher) handleFsEvent(event fsnotify.Event) { - // Get relative path - relPath, err := filepath.Rel(w.filesPath, event.Name) - if err != nil { - relPath = event.Name - } - - // Clean up path - relPath = filepath.ToSlash(relPath) - - // Determine event type - var eventType FileEventType - switch { - case event.Op&fsnotify.Create == fsnotify.Create: - eventType = EventTypeAdd - // If it's a new directory, add it to watch - if info, err := os.Stat(event.Name); err == nil && info.IsDir() { - w.fsWatcher.Add(event.Name) - } - case event.Op&fsnotify.Write == fsnotify.Write: - eventType = EventTypeChange - case event.Op&fsnotify.Remove == fsnotify.Remove: - eventType = EventTypeDelete - case event.Op&fsnotify.Rename == fsnotify.Rename: - eventType = EventTypeRename - default: - return // Ignore other events - } - - // Skip directories for file events (except remove) - if eventType != EventTypeDelete { - info, err := os.Stat(event.Name) - if err == nil && info.IsDir() { - return - } - } - - // Get file info if available - var size int64 - if eventType != EventTypeDelete { - if info, err := os.Stat(event.Name); err == nil { - size = info.Size() - } - } - - fileEvent := FileEvent{ - Type: eventType, - Path: relPath, - Size: size, - Timestamp: NowMS(), - } - - // Send to debouncer - w.debouncer.Add(fileEvent) -} - // handleDebouncedEvent processes a debounced event func (w *Watcher) handleDebouncedEvent(event FileEvent) { // Compute partial hash for add/change events @@ -224,44 +65,11 @@ func (w *Watcher) handleDebouncedEvent(event FileEvent) { // Dispatch to subscribers w.dispatcher.Dispatch(event) - - log.Printf("[Watcher] Event: %s %s", event.Type, event.Path) } -// RunScan performs a full directory scan -func (w *Watcher) RunScan() { - w.mu.Lock() - if w.isScanning { - w.mu.Unlock() - return - } - w.isScanning = true - w.mu.Unlock() - - defer func() { - w.mu.Lock() - w.isScanning = false - w.lastScan = NowMS() - w.mu.Unlock() - }() - - log.Println("[Watcher] Starting directory scan...") - - fileCount := 0 - for _, watchPath := range w.watchPaths { - count := w.scanDirectory(watchPath) - fileCount += count - } - - w.mu.Lock() - w.fileCount = fileCount - w.mu.Unlock() - - log.Printf("[Watcher] Scan complete: %d files found", fileCount) -} - -// scanDirectory scans a directory and emits add events for files -func (w *Watcher) scanDirectory(root string) int { +// ScanDirectory scans a directory and emits add events for files +// Returns the count of files found +func (w *Watcher) ScanDirectory(root string) int { count := 0 filepath.Walk(root, func(path string, info os.FileInfo, err error) error { @@ -286,7 +94,7 @@ func (w *Watcher) scanDirectory(root string) int { } relPath = filepath.ToSlash(relPath) - // Emit add event (through debouncer) + // Emit add event event := FileEvent{ Type: EventTypeAdd, Path: relPath, @@ -299,7 +107,7 @@ func (w *Watcher) scanDirectory(root string) int { event.PartialHash = hash } - // Dispatch directly (skip debouncer for scan) + // Dispatch directly w.dispatcher.Dispatch(event) count++ @@ -314,25 +122,34 @@ func (w *Watcher) scanDirectory(root string) int { func (w *Watcher) ScanMountPath(mountPath string) (int, error) { // Validate path is under filesPath if !strings.HasPrefix(mountPath, w.filesPath) { - return 0, fmt.Errorf("mount path %s is not under files path %s", mountPath, w.filesPath) + // Try to normalize paths + absMount, err := filepath.Abs(mountPath) + if err == nil { + absFiles, err := filepath.Abs(w.filesPath) + if err == nil && strings.HasPrefix(absMount, absFiles) { + mountPath = absMount + } + } } // Check directory exists info, err := os.Stat(mountPath) if err != nil { if os.IsNotExist(err) { - return 0, fmt.Errorf("mount path %s does not exist", mountPath) + log.Printf("[Watcher] Mount path %s does not exist, skipping", mountPath) + return 0, nil } - return 0, fmt.Errorf("failed to stat mount path %s: %w", mountPath, err) + return 0, err } if !info.IsDir() { - return 0, fmt.Errorf("mount path %s is not a directory", mountPath) + log.Printf("[Watcher] Mount path %s is not a directory, skipping", mountPath) + return 0, nil } // Scan the directory log.Printf("[Watcher] Scanning mount path: %s", mountPath) - count := w.scanDirectory(mountPath) + count := w.ScanDirectory(mountPath) log.Printf("[Watcher] Mount scan complete: %s (%d files)", mountPath, count) return count, nil @@ -356,22 +173,9 @@ func (w *Watcher) GetRecentEvents(sinceMS int64, limit int) []FileEvent { return result } -// GetStatus returns watcher status -func (w *Watcher) GetStatus() ScanStatusResponse { - w.mu.RLock() - defer w.mu.RUnlock() - - status := "running" - if !w.isRunning { - status = "stopped" - } - - return ScanStatusResponse{ - Status: status, - Scanning: w.isScanning, - LastScan: w.lastScan, - FileCount: w.fileCount, - } +// GetFilesPath returns the files path +func (w *Watcher) GetFilesPath() string { + return w.filesPath } // computePartialHash computes SHA-256 hash of first 64KB of a file diff --git a/internal/watchers/handlers.go b/internal/watchers/handlers.go new file mode 100644 index 0000000..4c59439 --- /dev/null +++ b/internal/watchers/handlers.go @@ -0,0 +1,192 @@ +package watchers + +import ( + "encoding/json" + "net/http" + + "github.com/gorilla/mux" +) + +// Handlers provides HTTP handlers for watchers management +type Handlers struct { + manager *Manager + poller *Poller +} + +// NewHandlers creates new watchers handlers +func NewHandlers(manager *Manager, poller *Poller) *Handlers { + return &Handlers{ + manager: manager, + poller: poller, + } +} + +// RegisterRoutes registers all watchers-related routes +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/{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") +} + +// handleList handles GET /api/watchers +func (h *Handlers) handleList(w http.ResponseWriter, r *http.Request) { + statuses := h.poller.GetWatcherStatuses() + + writeJSON(w, http.StatusOK, WatchersListResponse{ + Watchers: statuses, + Count: len(statuses), + }) +} + +// handleGet handles GET /api/watchers/{id} +func (h *Handlers) handleGet(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + id := vars["id"] + + watcher, err := h.manager.Get(id) + if err != nil { + writeError(w, http.StatusNotFound, err.Error()) + return + } + + // Get runtime status + status := h.poller.GetStatus(id) + + response := WatcherStatus{ + WatcherConfig: *watcher, + Active: status.Active, + LastScan: status.LastScan, + FileCount: status.FileCount, + IsScanning: status.IsScanning, + } + + writeJSON(w, http.StatusOK, response) +} + +// handleCreate handles POST /api/watchers +func (h *Handlers) handleCreate(w http.ResponseWriter, r *http.Request) { + var req CreateWatcherRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "invalid JSON body") + return + } + + watcher, err := h.manager.Create(req) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + + writeJSON(w, http.StatusCreated, watcher) +} + +// handleUpdate handles PUT /api/watchers/{id} +func (h *Handlers) handleUpdate(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + id := vars["id"] + + var req UpdateWatcherRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "invalid JSON body") + return + } + + watcher, err := h.manager.Update(id, req) + if err != nil { + if err.Error() == "watcher not found: "+id { + writeError(w, http.StatusNotFound, err.Error()) + } else { + writeError(w, http.StatusBadRequest, err.Error()) + } + return + } + + // Get runtime status + status := h.poller.GetStatus(id) + + response := WatcherStatus{ + WatcherConfig: *watcher, + Active: status.Active, + LastScan: status.LastScan, + FileCount: status.FileCount, + IsScanning: status.IsScanning, + } + + writeJSON(w, http.StatusOK, response) +} + +// handleDelete handles DELETE /api/watchers/{id} +func (h *Handlers) handleDelete(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + id := vars["id"] + + if err := h.manager.Delete(id); 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": "Watcher deleted", + }) +} + +// handleScan handles POST /api/watchers/{id}/scan +func (h *Handlers) handleScan(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + id := vars["id"] + + fileCount, err := h.poller.TriggerScan(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": "Scan triggered", + "fileCount": fileCount, + }) +} + +// handleScanAll handles POST /api/watchers/scan-all +func (h *Handlers) handleScanAll(w http.ResponseWriter, r *http.Request) { + totalCount, err := h.poller.TriggerScanAll() + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + + writeJSON(w, http.StatusOK, map[string]interface{}{ + "status": "ok", + "message": "All watchers scanned", + "totalFiles": totalCount, + }) +} + +// Helper functions + +func writeJSON(w http.ResponseWriter, status int, data interface{}) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + json.NewEncoder(w).Encode(data) +} + +func writeError(w http.ResponseWriter, status int, message string) { + writeJSON(w, status, map[string]string{ + "error": http.StatusText(status), + "message": message, + }) +} diff --git a/internal/watchers/manager.go b/internal/watchers/manager.go new file mode 100644 index 0000000..2f4d327 --- /dev/null +++ b/internal/watchers/manager.go @@ -0,0 +1,314 @@ +package watchers + +import ( + "encoding/json" + "fmt" + "log" + "os" + "path/filepath" + "sync" + + "github.com/google/uuid" + "github.com/metazla/meta-core/internal/config" +) + +// Manager handles CRUD operations for watchers configuration +type Manager struct { + config *config.Config + watchersFile string + + mu sync.RWMutex + watchers map[string]*WatcherConfig + + // Callback to notify poller of changes + onChanged func() +} + +// NewManager creates a new watchers manager +func NewManager(cfg *config.Config) (*Manager, error) { + watchersFile := cfg.WatchersFilePath() + + m := &Manager{ + config: cfg, + watchersFile: watchersFile, + watchers: make(map[string]*WatcherConfig), + } + + // Ensure directory exists + dir := filepath.Dir(watchersFile) + if err := os.MkdirAll(dir, 0755); err != nil { + return nil, fmt.Errorf("failed to create watchers directory: %w", err) + } + + // Load or migrate + if err := m.loadOrMigrate(); err != nil { + return nil, err + } + + return m, nil +} + +// SetOnChanged sets the callback for watcher changes +func (m *Manager) SetOnChanged(cb func()) { + m.mu.Lock() + defer m.mu.Unlock() + m.onChanged = cb +} + +// loadOrMigrate loads existing config or migrates from ENV +func (m *Manager) loadOrMigrate() error { + // Try to load existing file + if _, err := os.Stat(m.watchersFile); err == nil { + return m.load() + } + + // File doesn't exist, check for ENV migration + if len(m.config.WatchFolderList) > 0 { + log.Printf("[Watchers] Migrating from WATCH_FOLDER_LIST env var") + return m.migrateFromEnv() + } + + // No config, start empty + log.Printf("[Watchers] No existing config, starting with empty watchers list") + return nil +} + +// migrateFromEnv creates watchers from WATCH_FOLDER_LIST env var +func (m *Manager) migrateFromEnv() error { + m.mu.Lock() + defer m.mu.Unlock() + + for _, path := range m.config.WatchFolderList { + id := uuid.New().String()[:8] + watcher := &WatcherConfig{ + ID: id, + Path: path, + IntervalMs: DefaultIntervalMs, + Enabled: true, + CreatedAt: NowMS(), + } + m.watchers[id] = watcher + log.Printf("[Watchers] Migrated watcher: %s -> %s", id, path) + } + + // Save migrated config + return m.saveUnsafe() +} + +// load loads watchers from file +func (m *Manager) load() error { + m.mu.Lock() + defer m.mu.Unlock() + + data, err := os.ReadFile(m.watchersFile) + if err != nil { + return fmt.Errorf("failed to read watchers file: %w", err) + } + + var file WatchersFile + if err := json.Unmarshal(data, &file); err != nil { + return fmt.Errorf("failed to parse watchers file: %w", err) + } + + m.watchers = make(map[string]*WatcherConfig) + for i := range file.Watchers { + w := file.Watchers[i] + m.watchers[w.ID] = &w + } + + log.Printf("[Watchers] Loaded %d watchers from file", len(m.watchers)) + return nil +} + +// save persists watchers to file (thread-safe) +func (m *Manager) save() error { + m.mu.Lock() + defer m.mu.Unlock() + return m.saveUnsafe() +} + +// saveUnsafe persists watchers to file (must hold lock) +func (m *Manager) saveUnsafe() error { + watchers := make([]WatcherConfig, 0, len(m.watchers)) + for _, w := range m.watchers { + watchers = append(watchers, *w) + } + + file := WatchersFile{ + Version: 1, + Watchers: watchers, + } + + data, err := json.MarshalIndent(file, "", " ") + if err != nil { + return fmt.Errorf("failed to marshal watchers: %w", err) + } + + if err := os.WriteFile(m.watchersFile, data, 0644); err != nil { + return fmt.Errorf("failed to write watchers file: %w", err) + } + + return nil +} + +// notifyChanged calls the onChanged callback if set +func (m *Manager) notifyChanged() { + m.mu.RLock() + cb := m.onChanged + m.mu.RUnlock() + + if cb != nil { + go cb() + } +} + +// List returns all watchers +func (m *Manager) List() []WatcherConfig { + m.mu.RLock() + defer m.mu.RUnlock() + + result := make([]WatcherConfig, 0, len(m.watchers)) + for _, w := range m.watchers { + result = append(result, *w) + } + return result +} + +// Get returns a watcher by ID +func (m *Manager) Get(id string) (*WatcherConfig, error) { + m.mu.RLock() + defer m.mu.RUnlock() + + w, ok := m.watchers[id] + if !ok { + return nil, fmt.Errorf("watcher not found: %s", id) + } + return &WatcherConfig{ + ID: w.ID, + Path: w.Path, + IntervalMs: w.IntervalMs, + Enabled: w.Enabled, + CreatedAt: w.CreatedAt, + }, nil +} + +// Create adds a new watcher +func (m *Manager) Create(req CreateWatcherRequest) (*WatcherConfig, error) { + m.mu.Lock() + defer m.mu.Unlock() + + // Validate path + if req.Path == "" { + return nil, fmt.Errorf("path is required") + } + + // Set defaults + intervalMs := req.IntervalMs + if intervalMs < MinIntervalMs { + intervalMs = DefaultIntervalMs + } + + id := uuid.New().String()[:8] + watcher := &WatcherConfig{ + ID: id, + Path: req.Path, + IntervalMs: intervalMs, + Enabled: req.Enabled, + CreatedAt: NowMS(), + } + + m.watchers[id] = watcher + + if err := m.saveUnsafe(); err != nil { + delete(m.watchers, id) + return nil, err + } + + log.Printf("[Watchers] Created watcher: %s -> %s", id, req.Path) + + // Notify after releasing lock + go m.notifyChanged() + + return &WatcherConfig{ + ID: watcher.ID, + Path: watcher.Path, + IntervalMs: watcher.IntervalMs, + Enabled: watcher.Enabled, + CreatedAt: watcher.CreatedAt, + }, nil +} + +// Update modifies an existing watcher +func (m *Manager) Update(id string, req UpdateWatcherRequest) (*WatcherConfig, error) { + m.mu.Lock() + defer m.mu.Unlock() + + w, ok := m.watchers[id] + if !ok { + return nil, fmt.Errorf("watcher not found: %s", id) + } + + if req.Path != nil { + if *req.Path == "" { + return nil, fmt.Errorf("path cannot be empty") + } + w.Path = *req.Path + } + + if req.IntervalMs != nil { + interval := *req.IntervalMs + if interval < MinIntervalMs { + interval = MinIntervalMs + } + w.IntervalMs = interval + } + + if req.Enabled != nil { + w.Enabled = *req.Enabled + } + + if err := m.saveUnsafe(); err != nil { + return nil, err + } + + log.Printf("[Watchers] Updated watcher: %s", id) + + // Notify after releasing lock + go m.notifyChanged() + + return &WatcherConfig{ + ID: w.ID, + Path: w.Path, + IntervalMs: w.IntervalMs, + Enabled: w.Enabled, + CreatedAt: w.CreatedAt, + }, nil +} + +// Delete removes a watcher +func (m *Manager) Delete(id string) error { + m.mu.Lock() + defer m.mu.Unlock() + + if _, ok := m.watchers[id]; !ok { + return fmt.Errorf("watcher not found: %s", id) + } + + delete(m.watchers, id) + + if err := m.saveUnsafe(); err != nil { + return err + } + + log.Printf("[Watchers] Deleted watcher: %s", id) + + // Notify after releasing lock + go m.notifyChanged() + + return nil +} + +// GetFilesPath returns the files path from config +func (m *Manager) GetFilesPath() string { + return m.config.FilesPath +} diff --git a/internal/watchers/poller.go b/internal/watchers/poller.go new file mode 100644 index 0000000..e767a89 --- /dev/null +++ b/internal/watchers/poller.go @@ -0,0 +1,328 @@ +package watchers + +import ( + "log" + "sync" + "time" + + "github.com/metazla/meta-core/internal/watcher" +) + +// ScanFunc is the function type for scanning a directory +type ScanFunc func(path string) int + +// watcherPoller manages polling for a single watcher +type watcherPoller struct { + watcherID string + stopChan chan struct{} + stopped bool + mu sync.Mutex +} + +// RuntimeStatus holds the runtime state for a watcher +type RuntimeStatus struct { + Active bool + LastScan int64 + FileCount int + IsScanning bool +} + +// Poller manages per-watcher polling goroutines +type Poller struct { + manager *Manager + scanner *watcher.Watcher + dispatcher *watcher.Dispatcher + + pollers map[string]*watcherPoller // key: watcher ID + status map[string]*RuntimeStatus + + running bool + mu sync.RWMutex +} + +// NewPoller creates a new watchers poller +func NewPoller(manager *Manager, scanner *watcher.Watcher, dispatcher *watcher.Dispatcher) *Poller { + p := &Poller{ + manager: manager, + scanner: scanner, + dispatcher: dispatcher, + pollers: make(map[string]*watcherPoller), + status: make(map[string]*RuntimeStatus), + } + + // Register callback for watcher changes + manager.SetOnChanged(p.syncWithWatchers) + + return p +} + +// Start begins the polling service +func (p *Poller) Start() error { + p.mu.Lock() + defer p.mu.Unlock() + + if p.running { + return nil + } + p.running = true + + log.Println("[WatcherPoller] Starting file watcher polling service") + + // Initial sync with watcher configurations + go p.syncWithWatchers() + + return nil +} + +// Stop stops all polling goroutines +func (p *Poller) Stop() error { + p.mu.Lock() + defer p.mu.Unlock() + + if !p.running { + return nil + } + p.running = false + + log.Println("[WatcherPoller] Stopping file watcher polling service") + + // Stop all pollers + for id, poller := range p.pollers { + p.stopPoller(poller) + delete(p.pollers, id) + } + + return nil +} + +// SyncWithWatchers synchronizes pollers with current watcher configurations +func (p *Poller) syncWithWatchers() { + p.mu.RLock() + running := p.running + p.mu.RUnlock() + + if !running { + return + } + + // Get current watchers + watchers := p.manager.List() + + p.mu.Lock() + defer p.mu.Unlock() + + // Re-check running state after acquiring lock + if !p.running { + return + } + + // Track which watchers we've seen + seenWatchers := make(map[string]bool) + + for _, w := range watchers { + seenWatchers[w.ID] = true + + existingPoller, exists := p.pollers[w.ID] + + if w.Enabled { + // Calculate interval + interval := w.IntervalMs + if interval < MinIntervalMs { + interval = DefaultIntervalMs + } + + if !exists { + // Start new poller + log.Printf("[WatcherPoller] Starting poller for watcher %s (path: %s, interval: %dms)", w.ID, w.Path, interval) + p.startPollerForWatcher(w.ID, w.Path, interval) + } + } else if exists { + // Stop existing poller + log.Printf("[WatcherPoller] Stopping poller for watcher %s", w.ID) + p.stopPoller(existingPoller) + delete(p.pollers, w.ID) + } + } + + // Stop pollers for watchers that no longer exist + for id, poller := range p.pollers { + if !seenWatchers[id] { + log.Printf("[WatcherPoller] Stopping poller for removed watcher %s", id) + p.stopPoller(poller) + delete(p.pollers, id) + } + } +} + +// startPollerForWatcher starts a polling goroutine for a watcher +func (p *Poller) startPollerForWatcher(watcherID, path string, intervalMs int) { + poller := &watcherPoller{ + watcherID: watcherID, + stopChan: make(chan struct{}), + } + p.pollers[watcherID] = poller + + // Initialize status + if _, exists := p.status[watcherID]; !exists { + p.status[watcherID] = &RuntimeStatus{} + } + p.status[watcherID].Active = true + + go func() { + // Run initial scan immediately + p.runPollScan(watcherID, path) + + ticker := time.NewTicker(time.Duration(intervalMs) * time.Millisecond) + defer ticker.Stop() + + for { + select { + case <-poller.stopChan: + return + case <-ticker.C: + p.runPollScan(watcherID, path) + } + } + }() +} + +// stopPoller stops a watcher poller +func (p *Poller) stopPoller(poller *watcherPoller) { + poller.mu.Lock() + defer poller.mu.Unlock() + + if poller.stopped { + return + } + poller.stopped = true + close(poller.stopChan) + + // Update status + if status, exists := p.status[poller.watcherID]; exists { + status.Active = false + } +} + +// runPollScan executes a scan for a watcher +func (p *Poller) runPollScan(watcherID, path string) { + if p.scanner == nil { + return + } + + // Mark as scanning + p.mu.Lock() + if status, exists := p.status[watcherID]; exists { + status.IsScanning = true + } + p.mu.Unlock() + + // Perform scan + fileCount := p.scanner.ScanDirectory(path) + + // Update status + p.mu.Lock() + if status, exists := p.status[watcherID]; exists { + status.LastScan = NowMS() + status.FileCount = fileCount + status.IsScanning = false + } + p.mu.Unlock() + + log.Printf("[WatcherPoller] Poll scan complete for watcher %s: %d files", watcherID, fileCount) +} + +// TriggerScan triggers an immediate scan for a watcher +func (p *Poller) TriggerScan(watcherID string) (int, error) { + w, err := p.manager.Get(watcherID) + if err != nil { + return 0, err + } + + if p.scanner == nil { + return 0, nil + } + + // 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() + + // Perform scan + fileCount := p.scanner.ScanDirectory(w.Path) + + // Update status + p.mu.Lock() + if status, exists := p.status[watcherID]; exists { + status.LastScan = NowMS() + status.FileCount = fileCount + status.IsScanning = false + } + p.mu.Unlock() + + return fileCount, nil +} + +// TriggerScanAll triggers scans for all enabled watchers +func (p *Poller) TriggerScanAll() (int, error) { + watchers := p.manager.List() + totalCount := 0 + + for _, w := range watchers { + if w.Enabled { + count, err := p.TriggerScan(w.ID) + if err != nil { + log.Printf("[WatcherPoller] Failed to scan watcher %s: %v", w.ID, err) + continue + } + totalCount += count + } + } + + return totalCount, nil +} + +// GetStatus returns the runtime status for a watcher +func (p *Poller) GetStatus(watcherID string) *RuntimeStatus { + p.mu.RLock() + defer p.mu.RUnlock() + + if status, exists := p.status[watcherID]; exists { + return &RuntimeStatus{ + Active: status.Active, + LastScan: status.LastScan, + FileCount: status.FileCount, + IsScanning: status.IsScanning, + } + } + return &RuntimeStatus{} +} + +// GetWatcherStatuses returns all watchers with their runtime status +func (p *Poller) GetWatcherStatuses() []WatcherStatus { + watchers := p.manager.List() + result := make([]WatcherStatus, 0, len(watchers)) + + p.mu.RLock() + defer p.mu.RUnlock() + + for _, w := range watchers { + status := WatcherStatus{ + WatcherConfig: w, + } + + if rs, exists := p.status[w.ID]; exists { + status.Active = rs.Active + status.LastScan = rs.LastScan + status.FileCount = rs.FileCount + status.IsScanning = rs.IsScanning + } + + result = append(result, status) + } + + return result +} diff --git a/internal/watchers/types.go b/internal/watchers/types.go new file mode 100644 index 0000000..01e7d2a --- /dev/null +++ b/internal/watchers/types.go @@ -0,0 +1,59 @@ +package watchers + +import "time" + +const ( + // MinIntervalMs is the minimum polling interval (5 seconds) + MinIntervalMs = 5000 + // DefaultIntervalMs is the default polling interval (30 seconds) + DefaultIntervalMs = 30000 +) + +// WatcherConfig represents a watcher configuration stored in watchers.json +type WatcherConfig struct { + ID string `json:"id"` + Path string `json:"path"` // e.g., "/files/watch", "/files/movies" + IntervalMs int `json:"intervalMs"` // Min: 5000 (5s), Default: 30000 (30s) + Enabled bool `json:"enabled"` + CreatedAt int64 `json:"createdAt"` +} + +// WatcherStatus includes runtime status in API responses +type WatcherStatus struct { + WatcherConfig + Active bool `json:"active"` // Polling goroutine running + LastScan int64 `json:"lastScan"` // Unix ms timestamp + FileCount int `json:"fileCount"` // Files found in last scan + IsScanning bool `json:"isScanning"` // Currently scanning +} + +// WatchersFile is the structure of watchers.json +type WatchersFile struct { + Version int `json:"version"` + Watchers []WatcherConfig `json:"watchers"` +} + +// CreateWatcherRequest is the request body for creating a watcher +type CreateWatcherRequest struct { + Path string `json:"path"` + IntervalMs int `json:"intervalMs"` + Enabled bool `json:"enabled"` +} + +// UpdateWatcherRequest is the request body for updating a watcher +type UpdateWatcherRequest struct { + Path *string `json:"path,omitempty"` + IntervalMs *int `json:"intervalMs,omitempty"` + Enabled *bool `json:"enabled,omitempty"` +} + +// WatchersListResponse is the API response for listing watchers +type WatchersListResponse struct { + Watchers []WatcherStatus `json:"watchers"` + Count int `json:"count"` +} + +// NowMS returns current time in milliseconds +func NowMS() int64 { + return time.Now().UnixMilli() +}