goencode/internal/watcher/watcher.go

package watcher

import (
	"log"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"github.com/fsnotify/fsnotify"
	"goencode/internal/db"
	"goencode/internal/queue"
)

type Manager struct {
	watcher      *fsnotify.Watcher
	queueManager *queue.Manager
	timers       map[string]*time.Timer
	timersMu     sync.Mutex
	processChan  chan string
	stopChan     chan struct{}
}

func NewManager(qm *queue.Manager) (*Manager, error) {
	w, err := fsnotify.NewWatcher()
	if err != nil {
		return nil, err
	}

	m := &Manager{
		watcher:      w,
		queueManager: qm,
		timers:       make(map[string]*time.Timer),
		processChan:  make(chan string, 10000),
		stopChan:     make(chan struct{}),
	}
	
	for i := 0; i < 3; i++ {
		go m.processWorker()
	}
	
	return m, nil
}

func (m *Manager) Start() {
	m.Reload()
	go m.watchLoop()
}

func (m *Manager) Stop() {
	close(m.stopChan)
	m.watcher.Close()
}

func (m *Manager) Reload() {
	// Remove all existing watches
	for _, path := range m.watcher.WatchList() {
		m.watcher.Remove(path)
	}

	folders, err := db.GetWatchFolders()
	if err != nil {
		log.Printf("Watcher failed to get folders: %v", err)
		return
	}

	for _, f := range folders {
		if !f.Enabled {
			continue
		}
		if err := os.MkdirAll(f.FolderPath, 0755); err != nil {
			log.Printf("Failed to create watch folder %s: %v", f.FolderPath, err)
			continue
		}
		log.Printf("Watching and scanning %s", f.FolderPath)
		
		// Walk the directory to add all subdirectories to watcher and scan existing files
		filepath.Walk(f.FolderPath, func(path string, info os.FileInfo, err error) error {
			if err != nil {
				log.Printf("Error accessing path %s during scan: %v", path, err)
				return nil
			}
			if info.IsDir() {
				if err := m.watcher.Add(path); err != nil {
					log.Printf("Failed to watch %s: %v", path, err)
				}
			} else {
				// Process existing file asynchronously
				go m.handleEvent(path)
			}
			return nil
		})
	}
}

func (m *Manager) watchLoop() {
	for {
		select {
		case <-m.stopChan:
			return
		case event, ok := <-m.watcher.Events:
			if !ok {
				return
			}
			if event.Has(fsnotify.Write) || event.Has(fsnotify.Create) {
				// Check if the created event is a directory
				if event.Has(fsnotify.Create) {
					info, err := os.Stat(event.Name)
					if err == nil && info.IsDir() {
						filepath.Walk(event.Name, func(path string, info os.FileInfo, err error) error {
							if err != nil {
								return nil
							}
							if info.IsDir() {
								m.watcher.Add(path)
							} else {
								go m.handleEvent(path)
							}
							return nil
						})
					}
				}
				m.handleEvent(event.Name)
			}
		case err, ok := <-m.watcher.Errors:
			if !ok {
				return
			}
			log.Printf("Watcher error: %v", err)
		}
	}
}

func (m *Manager) processWorker() {
	for {
		select {
		case <-m.stopChan:
			return
		case path := <-m.processChan:
			go m.processFile(path)
		}
	}
}

func (m *Manager) handleEvent(filePath string) {
	m.timersMu.Lock()
	defer m.timersMu.Unlock()

	if t, exists := m.timers[filePath]; exists {
		t.Stop()
	}

	m.timers[filePath] = time.AfterFunc(2*time.Minute, func() {
		m.timersMu.Lock()
		delete(m.timers, filePath)
		m.timersMu.Unlock()
		
		select {
		case m.processChan <- filePath:
		default:
			log.Printf("Process queue full, dropping %s", filePath)
		}
	})
}

func (m *Manager) processFile(filePath string) {

	info, err := os.Stat(filePath)
	if err != nil || info.IsDir() {
		return // File removed or is a directory
	}

	// Make sure it's not a temp file
	if filepath.Ext(filePath) == ".tmp" {
		return
	}

	// Wait until the file is fully copied and stable
	var lastSize int64 = info.Size()
	var lastModTime time.Time = info.ModTime()
	var stableCount int

	for {
		time.Sleep(5 * time.Second)
		currentInfo, err := os.Stat(filePath)
		if err != nil {
			return // File was likely deleted during copying
		}

		if currentInfo.Size() == lastSize && currentInfo.ModTime().Equal(lastModTime) {
			stableCount++
			if stableCount >= 3 { // Stable for 15 seconds
				break
			}
		} else {
			stableCount = 0
			lastSize = currentInfo.Size()
			lastModTime = currentInfo.ModTime()
		}
	}

	// Re-check after waiting
	alreadyInQueue, err := db.IsFileAlreadyProcessedOrQueued(filePath)
	if err != nil {
		log.Printf("Error checking DB for %s: %v", filePath, err)
		return
	}
	if alreadyInQueue {
		return
	}

	// Find which watch folder it belongs to
	folders, err := db.GetWatchFolders()
	if err != nil {
		return
	}

	var match db.WatchFolder
	found := false
	for _, f := range folders {
		if !f.Enabled {
			continue
		}
		// Check if filePath is inside f.FolderPath
		cleanPath := filepath.Clean(filePath)
		folderPath := filepath.Clean(strings.TrimSpace(f.FolderPath))
		
		var isMatch bool
		if cleanPath == folderPath {
			isMatch = true
		} else if folderPath == string(os.PathSeparator) {
			isMatch = strings.HasPrefix(cleanPath, folderPath)
		} else {
			isMatch = strings.HasPrefix(cleanPath, folderPath+string(os.PathSeparator))
		}
		
		if isMatch {
			match = f
			found = true
			break
		}
	}

	if !found {
		return
	}

	err = db.AddJob(filePath, match.MediaType, 0, match.TargetResolution, match.CustomFFmpegFlags, info.Size())
	if err != nil {
		log.Printf("Failed to add job for %s: %v", filePath, err)
		return
	}

	log.Printf("Added job for %s", filePath)
	m.queueManager.NotifySSE("job_added", nil)
	m.queueManager.Trigger()
}