feat: phase 4 — periodic scanning (watch mode) with change detection

Add the scheduling layer and the watch command so gitflow can rescan
repositories on an interval and surface state changes.

Scheduler (internal/scheduler):
- Scheduler runs a callback immediately and then on a time.Ticker until
  the context is cancelled; interval <= 0 means a single run
- Graceful shutdown: cancellation is never reported as an error (checked
  on the first run and after every run), so Ctrl-C / SIGTERM exit cleanly
- A non-cancellation run error stops the loop and is propagated

Change detection (pkg/status):
- Changed(prev, next) compares snapshots per repository path — status,
  branch, detached flag, ahead/behind, stash count, and file counts —
  and returns the repositories whose observable state differs, including
  repositories that newly appear

Watch command (cmd/gitflow):
- newWatchCmd validates that --interval is positive, prompts for the
  parent directory on a TTY when --dir is absent, and drives the
  scheduler with ScanOnce
- Each frame clears the screen on a terminal (or prints a RFC3339 header
  when output is piped, so watch doubles as a lightweight logger) and
  renders through the presenter
- Footer shows changed repositories since the previous frame ("▲ name:
  status") and the next scan time; JSON format emits one document per
  frame for scripting
- SIGINT/SIGTERM handled via signalContext for a clean stop

Testing:
- Scheduler: single run, periodic repetition, stop-on-error, pre-cancelled
  context, and cancellation-during-run (no error reported)
- status.Changed: unchanged repos ignored; status, file-count, and
  newly-appeared repos detected

Verified: go build, go vet, go test -race, gofmt clean; manual watch
smoke test over a scratch directory with 1s interval — frames render,
SIGINT and SIGTERM both exit 0 with no orphaned processes.
This commit is contained in:
dimitar 2026-08-02 08:44:29 +02:00
parent 5fe5eb5af8
commit 0c9c07e8de
6 changed files with 307 additions and 0 deletions

View File

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

95
cmd/gitflow/watch.go Normal file
View File

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

View File

@ -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
}
}
}
}

View File

@ -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)
}
}

View File

@ -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()
}

View File

@ -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))
}
}