meta-core/internal/watchers/poller.go

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")

	// No startup reset. The file:events stream is bounded by MAXLEN
	// (api-mediated-access PR A) and SSE consumers resume from their
	// persisted Last-Event-ID, so a startup wipe would only orphan
	// consumer state without buying anything. The historical reason for
	// wiping — "in-memory state is empty, all files re-emit as add" —
	// stopped being true when HydrateStateFromStorage landed: state is
	// hydrated from Redis before this runs, so a startup EmitReset would
	// tell consumers to throw out their world without sending the
	// re-emission they need to rebuild it.
	//
	// Operator-triggered resets (POST /api/watchers/{id}/reset) still
	// work — that path goes through TriggerReset, which clears both the
	// stream AND the watcher's in-memory state, so the next scan tick
	// emits add events for every file.

	// 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 differential 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 differential scan
	result := p.scanner.ScanDirectoryDiff(path, watcherID)

	// Update status
	p.mu.Lock()
	if status, exists := p.status[watcherID]; exists {
		status.LastScan = NowMS()
		status.FileCount = result.Total
		status.IsScanning = false
	}
	p.mu.Unlock()

	// Only log if there were changes
	if result.Added > 0 || result.Changed > 0 || result.Deleted > 0 {
		log.Printf("[WatcherPoller] Diff scan for %s: +%d ~%d -%d (total: %d)",
			watcherID, result.Added, result.Changed, result.Deleted, result.Total)
	}
}

// TriggerScan triggers an immediate differential 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 differential scan
	result := p.scanner.ScanDirectoryDiff(w.Path, watcherID)

	// Update status
	p.mu.Lock()
	if status, exists := p.status[watcherID]; exists {
		status.LastScan = NowMS()
		status.FileCount = result.Total
		status.IsScanning = false
	}
	p.mu.Unlock()

	log.Printf("[WatcherPoller] Triggered scan for %s: +%d ~%d -%d (total: %d)",
		watcherID, result.Added, result.Changed, result.Deleted, result.Total)

	return result.Total, nil
}

// TriggerScanAll triggers scans for all enabled watchers
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
}

// TriggerReset performs a full reset for a watcher:
// 1. Clears the Redis stream and emits a reset event
// 2. Clears the watcher's state
// 3. Performs a full scan (all files are "new")
func (p *Poller) TriggerReset(watcherID string) (int, error) {
	w, err := p.manager.Get(watcherID)
	if err != nil {
		return 0, err
	}

	if p.scanner == nil || p.dispatcher == nil {
		return 0, nil
	}

	log.Printf("[WatcherPoller] Triggering reset for watcher %s", watcherID)

	// Mark as scanning
	p.mu.Lock()
	if status, exists := p.status[watcherID]; exists {
		status.IsScanning = true
	} else {
		p.status[watcherID] = &RuntimeStatus{IsScanning: true}
	}
	p.mu.Unlock()

	// Step 1: Emit reset event (this clears the stream first)
	if err := p.dispatcher.EmitReset(watcherID); err != nil {
		log.Printf("[WatcherPoller] Warning: failed to emit reset for %s: %v", watcherID, err)
	}

	// Step 2: Clear the watcher's state
	p.scanner.ClearState(watcherID)

	// Step 3: Clear event buffer
	p.scanner.ClearEventBuffer()

	// Step 4: Perform full scan (all files are now "new" since state is cleared)
	result := p.scanner.ScanDirectoryDiff(w.Path, watcherID)

	// Update status
	p.mu.Lock()
	if status, exists := p.status[watcherID]; exists {
		status.LastScan = NowMS()
		status.FileCount = result.Total
		status.IsScanning = false
	}
	p.mu.Unlock()

	log.Printf("[WatcherPoller] Reset complete for %s: %d files", watcherID, result.Total)

	return result.Total, nil
}

// TriggerResetAll performs a full reset for all enabled watchers
func (p *Poller) TriggerResetAll() (int, error) {
	watchers := p.manager.List()
	totalCount := 0
	isFirst := true

	for _, w := range watchers {
		if w.Enabled {
			// Only emit reset event once (for the first watcher)
			// Subsequent watchers just clear state and rescan
			if isFirst {
				count, err := p.TriggerReset(w.ID)
				if err != nil {
					log.Printf("[WatcherPoller] Failed to reset watcher %s: %v", w.ID, err)
					continue
				}
				totalCount += count
				isFirst = false
			} else {
				// Clear state and rescan without clearing stream again
				p.scanner.ClearState(w.ID)
				result := p.scanner.ScanDirectoryDiff(w.Path, w.ID)
				totalCount += result.Total
				log.Printf("[WatcherPoller] Reset scan for %s: %d files", w.ID, result.Total)
			}
		}
	}

	return totalCount, nil
}

// GetStatus returns the runtime status for a watcher
func (p *Poller) GetStatus(watcherID string) *RuntimeStatus {
	p.mu.RLock()
	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
}