merge: phase 4 scheduler into main

cmd/gitflow/root.go

@@ -26,6 +26,7 @@ repositories that need attention.`,
 	}
 	root.AddCommand(
 		newScanCmd(),
+		newWatchCmd(),
 		newConfigCmd(),
 		newVersionCmd(),
 	)

cmd/gitflow/watch.go

@@ -0,0 +1,95 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"os"
+	"time"
+
+	"github.com/spf13/cobra"
+
+	"gitea.oblak.solutions/dimitar/gitFlow/internal/app"
+	"gitea.oblak.solutions/dimitar/gitFlow/internal/config"
+	"gitea.oblak.solutions/dimitar/gitFlow/internal/presenter"
+	"gitea.oblak.solutions/dimitar/gitFlow/internal/scheduler"
+	"gitea.oblak.solutions/dimitar/gitFlow/pkg/status"
+)
+
+// newWatchCmd repeatedly scans on an interval until interrupted.
+func newWatchCmd() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "watch",
+		Short: "Repeatedly scan repositories on an interval",
+		Args:  cobra.NoArgs,
+		RunE: func(cmd *cobra.Command, _ []string) error {
+			cfg, err := config.Load(cmd.Flags())
+			if err != nil {
+				return err
+			}
+			if cfg.Interval <= 0 {
+				return errors.New("watch requires a positive --interval (e.g. --interval 30s)")
+			}
+			// Same interactive prompt as scan, per the README.
+			if !cmd.Flags().Changed("dir") && isTerminal(os.Stdin) {
+				if d, err := promptDir(cfg.Dir); err == nil {
+					cfg.Dir = d
+				}
+			}
+
+			ctx, stop := signalContext()
+			defer stop()
+
+			a, err := app.New(cfg)
+			if err != nil {
+				return err
+			}
+			opts := presenter.Options{Color: presenter.ParseColorMode(cfg.Color)}
+
+			var prev status.ScanResult
+			first := true
+			run := func(ctx context.Context) error {
+				result, err := a.ScanOnce(ctx)
+				if err != nil {
+					return err
+				}
+				if cfg.Format == "json" {
+					return presenter.Present(os.Stdout, cfg.Format, result, opts)
+				}
+				renderWatchFrame(result, prev, first, cfg, opts)
+				prev = result
+				first = false
+				return nil
+			}
+
+			sched := scheduler.New(cfg.Interval, run)
+			return sched.Run(ctx)
+		},
+	}
+	config.RegisterFlags(cmd.Flags())
+	return cmd
+}
+
+// renderWatchFrame clears the screen (or prints a timestamp header when
+// output is not a terminal), renders the scan, and prints a footer with
+// changed repositories and the next scan time.
+func renderWatchFrame(result, prev status.ScanResult, first bool, cfg *config.Config, opts presenter.Options) {
+	if isTerminal(os.Stdout) {
+		fmt.Fprint(os.Stdout, "\x1b[2J\x1b[H")
+	} else {
+		fmt.Fprintf(os.Stdout, "--- %s ---\n", time.Now().Format(time.RFC3339))
+	}
+
+	if err := presenter.Present(os.Stdout, cfg.Format, result, opts); err != nil {
+		fmt.Fprintf(os.Stderr, "render: %v\n", err)
+		return
+	}
+
+	if !first {
+		for _, r := range status.Changed(prev, result) {
+			fmt.Fprintf(os.Stdout, "▲ %s: %s\n", r.Name, r.Status)
+		}
+	}
+	fmt.Fprintf(os.Stdout, "watching %s every %s — next scan at %s (Ctrl-C to stop)\n",
+		cfg.Dir, cfg.Interval, time.Now().Add(cfg.Interval).Format("15:04:05"))
+}

internal/scheduler/scheduler.go

@@ -0,0 +1,55 @@
+// Package scheduler runs a function on a fixed interval with an immediate
+// first execution, until the context is cancelled.
+package scheduler
+
+import (
+	"context"
+	"time"
+)
+
+// Scheduler triggers a run function periodically.
+type Scheduler struct {
+	interval time.Duration
+	run      func(ctx context.Context) error
+}
+
+// New creates a Scheduler that calls run immediately and then every
+// interval. An interval <= 0 means run once and stop.
+func New(interval time.Duration, run func(ctx context.Context) error) *Scheduler {
+	return &Scheduler{interval: interval, run: run}
+}
+
+// Run executes run immediately and then every interval until ctx is
+// cancelled. A run error stops the loop and is returned, except when the
+// context was cancelled (graceful shutdown is not an error).
+func (s *Scheduler) Run(ctx context.Context) error {
+	if err := ctx.Err(); err != nil {
+		return nil
+	}
+	if err := s.run(ctx); err != nil {
+		if ctx.Err() != nil {
+			return nil
+		}
+		return err
+	}
+	if s.interval <= 0 {
+		return nil
+	}
+
+	ticker := time.NewTicker(s.interval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-ticker.C:
+			if err := s.run(ctx); err != nil {
+				if ctx.Err() != nil {
+					return nil
+				}
+				return err
+			}
+		}
+	}
+}

internal/scheduler/scheduler_test.go

@@ -0,0 +1,95 @@
+package scheduler
+
+import (
+	"context"
+	"errors"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestRunOnce(t *testing.T) {
+	var calls atomic.Int32
+	s := New(0, func(ctx context.Context) error {
+		calls.Add(1)
+		return nil
+	})
+	if err := s.Run(context.Background()); err != nil {
+		t.Fatalf("Run: %v", err)
+	}
+	if got := calls.Load(); got != 1 {
+		t.Errorf("run called %d times, want 1", got)
+	}
+}
+
+func TestRunPeriodic(t *testing.T) {
+	var calls atomic.Int32
+	s := New(10*time.Millisecond, func(ctx context.Context) error {
+		calls.Add(1)
+		return nil
+	})
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- s.Run(ctx) }()
+
+	time.Sleep(55 * time.Millisecond)
+	cancel()
+	if err := <-done; err != nil {
+		t.Fatalf("Run: %v", err)
+	}
+	if got := calls.Load(); got < 3 {
+		t.Errorf("run called %d times, want >= 3", got)
+	}
+}
+
+func TestRunStopsOnError(t *testing.T) {
+	var calls atomic.Int32
+	wantErr := errors.New("boom")
+	s := New(5*time.Millisecond, func(ctx context.Context) error {
+		n := calls.Add(1)
+		if n == 2 {
+			return wantErr
+		}
+		return nil
+	})
+	if err := s.Run(context.Background()); !errors.Is(err, wantErr) {
+		t.Fatalf("Run() error = %v, want %v", err, wantErr)
+	}
+	if got := calls.Load(); got != 2 {
+		t.Errorf("run called %d times, want 2", got)
+	}
+}
+
+func TestRunCancelledBeforeStart(t *testing.T) {
+	var calls atomic.Int32
+	s := New(time.Second, func(ctx context.Context) error {
+		calls.Add(1)
+		return nil
+	})
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	if err := s.Run(ctx); err != nil {
+		t.Fatalf("Run: %v", err)
+	}
+	if got := calls.Load(); got != 0 {
+		t.Errorf("run called %d times, want 0", got)
+	}
+}
+
+func TestRunCancellationIsNotAnError(t *testing.T) {
+	var calls atomic.Int32
+	s := New(5*time.Millisecond, func(ctx context.Context) error {
+		calls.Add(1)
+		time.Sleep(50 * time.Millisecond) // outlive cancellation
+		return ctx.Err()
+	})
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- s.Run(ctx) }()
+
+	time.Sleep(12 * time.Millisecond)
+	cancel()
+	if err := <-done; err != nil {
+		t.Fatalf("Run returned %v on cancellation, want nil", err)
+	}
+}

pkg/status/status.go

@@ -167,3 +167,33 @@ func (r ScanResult) NeedsAttention() []RepoInfo {
 	}
 	return out
 }
+
+// Changed reports the repositories whose observable state differs between
+// prev and next, in next's order. Repositories that appear only in next
+// (or vanished from prev) count as changed.
+func Changed(prev, next ScanResult) []RepoInfo {
+	byPath := make(map[string]RepoInfo, len(prev.Repos))
+	for _, r := range prev.Repos {
+		byPath[r.Path] = r
+	}
+	out := make([]RepoInfo, 0, len(next.Repos))
+	for _, r := range next.Repos {
+		p, ok := byPath[r.Path]
+		if !ok || !sameState(p, r) {
+			out = append(out, r)
+		}
+	}
+	return out
+}
+
+// sameState reports whether two snapshots of the same repository are
+// observably identical.
+func sameState(a, b RepoInfo) bool {
+	return a.Status == b.Status &&
+		a.Branch == b.Branch &&
+		a.Detached == b.Detached &&
+		a.AheadBy == b.AheadBy &&
+		a.BehindBy == b.BehindBy &&
+		a.StashCount == b.StashCount &&
+		a.FileCount() == b.FileCount()
+}

pkg/status/status_test.go

@@ -118,3 +118,34 @@ func TestScanResultNeedsAttention(t *testing.T) {
 		}
 	}
 }
+
+func TestChanged(t *testing.T) {
+	prev := ScanResult{Repos: []RepoInfo{
+		{Path: "/a", Status: StatusClean},
+		{Path: "/b", Status: StatusModified, ModifiedFiles: []string{"x"}},
+		{Path: "/c", Status: StatusAhead, AheadBy: 2},
+	}}
+	next := ScanResult{Repos: []RepoInfo{
+		{Path: "/a", Status: StatusClean}, // unchanged
+		{Path: "/b", Status: StatusModified, ModifiedFiles: []string{"x"}, UntrackedFiles: []string{"y"}}, // file count changed
+		{Path: "/c", Status: StatusClean},               // status changed
+		{Path: "/d", Status: StatusBehind, BehindBy: 3}, // newly appeared
+	}}
+
+	got := Changed(prev, next)
+	byPath := make(map[string]bool, len(got))
+	for _, r := range got {
+		byPath[r.Path] = true
+	}
+	if byPath["/a"] {
+		t.Error("/a reported as changed but state is identical")
+	}
+	for _, want := range []string{"/b", "/c", "/d"} {
+		if !byPath[want] {
+			t.Errorf("%s not reported as changed", want)
+		}
+	}
+	if len(got) != 3 {
+		t.Errorf("Changed() returned %d repos, want 3", len(got))
+	}
+}