scan: add --every flag to support periodic scanning

Author: rubiojr

count.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"fmt"
 	"os"
 	"path/filepath"
@@ -13,7 +14,7 @@ type FileCount struct {
 	Errors []error
 }
 
-func FileCounter(root string) *FileCount {
+func FileCounter(ctx context.Context, root string) *FileCount {
 	count := &FileCount{
 		Chan: make(chan int64),
 	}
@@ -26,6 +27,10 @@ func FileCounter(root string) *FileCount {
 				log.Printf("error walking path %s: %v", path, err)
 				return nil
 			}
+
+			if ctx.Err() != nil {
+				return ctx.Err()
+			}
 			counter++
 
 			if info.IsDir() {

internal/cache/cache.go

@@ -104,7 +104,7 @@ func (fc *FileCache) Save() error {
 		return fmt.Errorf("failed to create cache directory: %v", err)
 	}
 
-	log.Printf("Saving file cache to %s", fc.cachePath)
+	log.Debugf("Saving file cache to %s", fc.cachePath)
 	return fc.cache.SaveToFile(fc.cachePath)
 }
 

internal/processors/nats/nats.go

@@ -168,7 +168,6 @@ func (np *natsProcessor) Process(path string, msg types.ScannedFile) error {
 			return fmt.Errorf("failed to encrypt message: %v", err)
 		}
 		publishData = encryptedData
-		log.Debugf("Message encrypted: %d bytes -> %d bytes", len(plainData), len(publishData))
 	} else {
 		publishData = plainData
 	}

internal/scanner/scanner.go

@@ -125,8 +125,13 @@ func (s *DirectoryScanner) incCounter() {
 }
 
 func (s *DirectoryScanner) ScanDirectory(ctx context.Context, processor processors.Processor) (int64, error) {
-	defer s.pool.Stop()
-	defer s.cache.Save()
+	defer func() {
+		s.pool.Stop()
+		err := s.cache.Save()
+		if err != nil {
+			log.Errorf("Error saving cache: %v", err)
+		}
+	}()
 
 	hostname, err := os.Hostname()
 	if err != nil {
@@ -140,28 +145,35 @@ func (s *DirectoryScanner) ScanDirectory(ctx context.Context, processor processo
 		if ctx.Err() != nil {
 			return ctx.Err()
 		}
+
+		if err != nil {
+			log.Errorf("Error accessing %q: %v", path, err)
+			return err
+		}
+
+		count++
 		s.incCounter()
 
 		if s.ignoreHidden && info.IsDir() && len(info.Name()) > 1 && info.Name()[0] == '.' {
-			log.Printf("ignoring hidden directory: %s", path)
+			log.Debugf("ignoring hidden directory: %s", path)
 			return filepath.SkipDir
 		}
 
 		if s.ignoreHidden && info.Name()[0] == '.' {
-			log.Printf("ignoring hidden file: %s", path)
+			log.Debugf("ignoring hidden file: %s", path)
 			return nil
 		}
 
 		// Skip files that cannot be accessed.
 		absPath, err := filepath.Abs(path)
 		if err != nil {
-			log.Printf("Error accessing %q: %v", path, err)
+			log.Debugf("Error accessing %q: %v", path, err)
 			return nil
 		}
 
 		// Skip ignored directories
 		if info.IsDir() && slices.Contains(ignoredDirectories, info.Name()) {
-			log.Printf("ignoring directory %s", path)
+			log.Debugf("ignoring directory %s", path)
 			return filepath.SkipDir
 		}
 
@@ -172,21 +184,20 @@ func (s *DirectoryScanner) ScanDirectory(ctx context.Context, processor processo
 
 		// return if the file is not a regular file
 		if !info.Mode().IsRegular() {
-			//log.Printf("Warn: not a regular file %q", path)
 			return nil
 		}
 
 		// Skip ignored files
 		if slices.Contains(ignoredFiles, info.Name()) {
-			log.Printf("ignoring file %s", path)
+			log.Debugf("ignoring file %s", path)
 			return nil
 		}
 
 		// ignoreList is a list of regular expressions to ignore
 		for _, pattern := range s.ignoreList {
 			matched, _ := regexp.MatchString(pattern, absPath)
 			if matched {
-				log.Printf("ignoring path match %s", path)
+				log.Debugf("ignoring path match %s", path)
 				if info.IsDir() {
 					return filepath.SkipDir
 				}
@@ -226,18 +237,15 @@ func (s *DirectoryScanner) ScanDirectory(ctx context.Context, processor processo
 				Hostname:  hostname,
 			}
 
+			log.Debugf("Processing file %s\n", absPath)
 			err = processor.Process(absPath, msg)
 			if err != nil {
-				if ctx.Err() != nil {
-					return ctx.Err()
-				}
-				log.Errorf("failed processing %q: %v", path, err)
+				log.Errorf("failed processing %q: %v", absPath, err)
 			}
+			log.Debugf("Marking file %s processed\n", absPath)
 			s.cache.MarkFileProcessed(absPath, fileHash)
 			return nil
 		}
-
-		count++
 		s.pool.Submit(f)
 		return nil
 	})

main.go

@@ -68,11 +68,18 @@ func main() {
 						Name:  "ca-cert",
 						Usage: "TLS CA cert",
 					},
+					&cli.StringFlag{
+						Name:  "every",
+						Usage: "Run the scanner regularly. Interval specified in seconds(s), minutes(m) or hours(h)",
+					},
 				},
 				Action: func(c *cli.Context) error {
 					if c.Bool("debug") {
 						os.Setenv("HASHUP_DEBUG", "1")
 					}
+					if c.String("every") != "" {
+						return runEvery(c)
+					}
 					return runScanner(c)
 				},
 			},

scanner.go

@@ -6,10 +6,8 @@ import (
 	"fmt"
 	"io"
 	"os"
-	"os/signal"
 	"path/filepath"
 	"strings"
-	"syscall"
 	"time"
 
 	"github.com/rubiojr/hashup/internal/cache"
@@ -20,11 +18,34 @@ import (
 	"github.com/urfave/cli/v2"
 )
 
+func runEvery(c *cli.Context) error {
+	d, err := time.ParseDuration(c.String("every"))
+	if err != nil {
+		return fmt.Errorf("failed to parse duration: %v", err)
+	}
+
+	ticker := time.NewTicker(d)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			err := runScanner(c)
+			if err != nil {
+				log.Errorf("failed to run scanner: %v", err)
+			}
+		case <-c.Context.Done():
+			return c.Context.Err()
+		}
+	}
+}
+
 func runScanner(clictx *cli.Context) error {
 	cfg, err := config.LoadConfigFromCLI(clictx)
 	if err != nil {
 		return fmt.Errorf("failed to load config: %v", err)
 	}
+
 	encryptionKey := cfg.Main.EncryptionKey
 	if encryptionKey == "" {
 		return fmt.Errorf("encryption key is required")
@@ -44,15 +65,6 @@ func runScanner(clictx *cli.Context) error {
 		log.SetOutput(io.Discard)
 	}
 
-	var fileCount int64
-	// Count and print the number of files to be indexed
-	go func() {
-		tnow := time.Now()
-		fileCount = <-FileCounter(rootDir).Chan
-		elapsed := time.Since(tnow)
-		log.Printf("Counted %d files in %s\n", fileCount, elapsed)
-	}()
-
 	var ignoreList []string
 	if clictx.String("ignore-file") != "" {
 		var err error
@@ -62,6 +74,15 @@ func runScanner(clictx *cli.Context) error {
 		}
 	}
 
+	var fileCount int64
+	// Count and print the number of files to be indexed
+	go func() {
+		tnow := time.Now()
+		fileCount = <-FileCounter(clictx.Context, rootDir).Chan
+		elapsed := time.Since(tnow)
+		log.Debugf("Counted %d files in %s\n", fileCount, elapsed)
+	}()
+
 	scannerOpts := []scanner.Option{
 		scanner.WithIgnoreList(ignoreList),
 		scanner.WithIgnoreHidden(clictx.Bool("ignore-hidden")),
@@ -70,14 +91,20 @@ func runScanner(clictx *cli.Context) error {
 	scanner := scanner.NewDirectoryScanner(rootDir, scannerOpts...)
 
 	var pCounter int64
+	counterChan := scanner.CounterChan()
 	go func() {
-		for range scanner.CounterChan() {
-			pCounter++
-			if fileCount != 0 {
-				percent := float64(pCounter) / float64(fileCount) * 100
-				fmt.Printf("Scanned [%d/%d] files (%.0f%%)\r", pCounter, fileCount, percent)
-			} else {
-				fmt.Printf("Scanned %d files\r", pCounter)
+		for {
+			select {
+			case <-clictx.Done():
+				return
+			case <-counterChan:
+				pCounter++
+				if fileCount != 0 {
+					percent := float64(pCounter) / float64(fileCount) * 100
+					fmt.Printf("Scanned [%d/%d] files (%.0f%%)\r", pCounter, fileCount, percent)
+				} else {
+					fmt.Printf("Scanned %d files\r", pCounter)
+				}
 			}
 		}
 	}()
@@ -102,14 +129,19 @@ func runScanner(clictx *cli.Context) error {
 	}
 
 	go func() {
-		for stats := range statsChan {
-			processedFiles++
-			skippedFiles += int64(stats.SkippedFiles)
-			queuedFiles += int64(stats.QueuedFiles)
+		for {
+			select {
+			case <-clictx.Done():
+				return
+			case stats := <-statsChan:
+				processedFiles++
+				skippedFiles += int64(stats.SkippedFiles)
+				queuedFiles += int64(stats.QueuedFiles)
+			}
 		}
 	}()
 
-	ctx, cancel := context.WithCancel(context.Background())
+	ctx, cancel := context.WithCancel(clictx.Context)
 	defer cancel()
 	processor, err := nats.NewNATSProcessor(
 		ctx,
@@ -124,7 +156,7 @@ func runScanner(clictx *cli.Context) error {
 	}
 	defer processor.Close()
 
-	signalCh := make(chan os.Signal, 1)
+	done := make(chan bool)
 	go func() {
 		startTime := time.Now()
 		fmt.Printf("Starting directory scan in %s...\n", rootDir)
@@ -134,21 +166,28 @@ func runScanner(clictx *cli.Context) error {
 			log.Errorf("error scanning directory: %v", err)
 		}
 		elapsed := time.Since(startTime)
-		fmt.Printf("Completed scanning %d files in %q in %v\r", count, rootDir, elapsed)
-		cancel()
-		signalCh <- syscall.SIGINT
+		fmt.Printf("Completed scanning %d files in %q in %v\r\n", count, rootDir, elapsed)
+		done <- true
 	}()
 
-	signal.Notify(signalCh, syscall.SIGINT, syscall.SIGTERM)
-	<-signalCh
-	log.Printf("Shutting down...")
-	cancel()
+Loop:
+	for {
+		select {
+		case <-done:
+			log.Printf("Shutting down...")
+			break Loop
+		case <-ctx.Done():
+			log.Printf("Context canceled")
+			break Loop
+		}
+	}
 	fmt.Printf(
 		"Processed %d files, skipped %d files, queued %d files\n",
 		processedFiles,
 		skippedFiles,
 		queuedFiles,
 	)
+
 	return nil
 }