//go:build unix

package hostfiles

import (
	"context"
	"errors"
	"log"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/release"
)

type fakeWatcher struct {
	ch        chan event
	mu        sync.Mutex
	degraded  bool
	repairs   int
	repairErr error // when nil, Repair succeeds and clears degraded
}

func newFake() *fakeWatcher { return &fakeWatcher{ch: make(chan event, 64)} }

func (f *fakeWatcher) Events() <-chan event { return f.ch }
func (f *fakeWatcher) Degraded() bool       { f.mu.Lock(); defer f.mu.Unlock(); return f.degraded }
func (f *fakeWatcher) Close() error         { return nil }
func (f *fakeWatcher) Repair() error {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.repairs++
	if f.repairErr == nil {
		f.degraded = false
	}
	return f.repairErr
}
func (f *fakeWatcher) setDegraded(v bool) { f.mu.Lock(); f.degraded = v; f.mu.Unlock() }

// eventsOnly publishes on events and nothing else: no repair scan and no
// polling can satisfy a test that uses it. polled is the opposite.
var (
	eventsOnly = Options{Deadline: 40 * time.Millisecond, Scan: time.Hour, Poll: time.Hour}
	polled     = Options{Deadline: 40 * time.Millisecond, Scan: time.Hour, Poll: 40 * time.Millisecond}
)

// startRun runs the loop and returns once its first full scan has
// finished, so nothing a test does afterwards can be found by that scan.
func startRun(t *testing.T, p *Publisher, dir string, w watcher, o Options) (stop func() error) {
	t.Helper()
	first := make(chan struct{})
	var once sync.Once
	o.OnStats = func(SyncStats) { once.Do(func() { close(first) }) }
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	go func() { done <- run(ctx, p, dir, o, w) }()
	select {
	case <-first:
	case err := <-done:
		t.Fatalf("run ended before its first scan: %v", err)
	case <-time.After(2 * time.Second):
		t.Fatal("run never finished its first scan")
	}
	return func() error { cancel(); return <-done }
}

// startRunCounting is startRun with a callback on every full scan.
func startRunCounting(t *testing.T, p *Publisher, dir string, w watcher, o Options, onScan func()) (stop func() error) {
	t.Helper()
	first := make(chan struct{})
	var once sync.Once
	o.OnStats = func(SyncStats) { onScan(); once.Do(func() { close(first) }) }
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	go func() { done <- run(ctx, p, dir, o, w) }()
	select {
	case <-first:
	case <-time.After(2 * time.Second):
		t.Fatal("run never finished its first scan")
	}
	return func() error { cancel(); return <-done }
}

// eventually polls until ok returns true or two seconds pass.
func eventually(t *testing.T, what string, ok func() bool) {
	t.Helper()
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		if ok() {
			return
		}
		time.Sleep(10 * time.Millisecond)
	}
	t.Fatalf("timed out waiting for: %s", what)
}

func manifestHas(s release.Store, path, sha string) bool {
	m, found, err := Load(context.Background(), s, pfx)
	if err != nil || !found {
		return false
	}
	e, ok := m.Entries[path]
	return ok && !e.Deleted && e.SHA256 == sha
}

func TestRunPublishesATouchedFileAfterTheDeadline(t *testing.T) {
	s, dir, w := release.NewDirStore(t.TempDir()), t.TempDir(), newFake()
	p := open(t, s)
	stop := startRun(t, p, dir, w, eventsOnly)
	defer stop()
	write(t, dir, "instance.key", "k1")
	time.Sleep(3 * eventsOnly.Deadline)
	if manifestHas(s, "instance.key", sum([]byte("k1"))) {
		t.Fatal("published with no event: something other than the event path found it")
	}
	touched := time.Now()
	w.ch <- event{rel: "instance.key"}
	eventually(t, "instance.key published", func() bool { return manifestHas(s, "instance.key", sum([]byte("k1"))) })
	if took := time.Since(touched); took < eventsOnly.Deadline/2 {
		t.Fatalf("published %s after the event; the deadline is %s", took, eventsOnly.Deadline)
	}
}

func TestRunPublishesADeleteAsATombstone(t *testing.T) {
	s, dir, w := release.NewDirStore(t.TempDir()), t.TempDir(), newFake()
	p := open(t, s)
	write(t, dir, "a", "a")
	stop := startRun(t, p, dir, w, eventsOnly) // the first scan publishes "a"
	defer stop()
	eventually(t, "a published", func() bool { return manifestHas(s, "a", sum([]byte("a"))) })
	if err := os.Remove(filepath.Join(dir, "a")); err != nil {
		t.Fatal(err)
	}
	w.ch <- event{rel: "a"}
	eventually(t, "a tombstoned", func() bool {
		m, _, _ := Load(context.Background(), s, pfx)
		return m.Entries["a"].Deleted
	})
}

func TestRunRescansOnOverflow(t *testing.T) {
	s, dir, w := release.NewDirStore(t.TempDir()), t.TempDir(), newFake()
	p := open(t, s)
	stop := startRun(t, p, dir, w, eventsOnly)
	defer stop()
	write(t, dir, "missed.txt", "m") // no event for it: the queue overflowed
	time.Sleep(3 * eventsOnly.Deadline)
	if manifestHas(s, "missed.txt", sum([]byte("m"))) {
		t.Fatal("found before the rescan was asked for")
	}
	w.ch <- event{rescan: true}
	eventually(t, "missed.txt found by the rescan", func() bool { return manifestHas(s, "missed.txt", sum([]byte("m"))) })
}

// A2-5 residual: with a watch missing, changes are found by polling, not
// by waiting for the ten-minute scan.
func TestRunPollsWhileTheWatcherIsDegradedAndKeepsTryingToRepair(t *testing.T) {
	s, dir, w := release.NewDirStore(t.TempDir()), t.TempDir(), newFake()
	w.setDegraded(true)
	w.repairErr = errors.New("watch limit reached")
	p := open(t, s)
	stop := startRun(t, p, dir, w, polled)
	defer stop()
	write(t, dir, "quiet.txt", "q") // no event at all
	eventually(t, "quiet.txt found by polling", func() bool { return manifestHas(s, "quiet.txt", sum([]byte("q"))) })
	eventually(t, "more than one repair attempt", func() bool { w.mu.Lock(); defer w.mu.Unlock(); return w.repairs > 1 })
}

// The other half: once repair succeeds, polling stops. The loop runs
// with a short polling period, so if it kept polling the scan count would
// keep climbing; it must scan once more for what changed meanwhile and
// then go quiet.
func TestRunScansOnceAfterARepairThenStopsPolling(t *testing.T) {
	s, dir, w := release.NewDirStore(t.TempDir()), t.TempDir(), newFake()
	w.setDegraded(true)
	w.repairErr = errors.New("not yet")
	p := open(t, s)
	scans := 0
	var mu sync.Mutex
	count := func() int { mu.Lock(); defer mu.Unlock(); return scans }
	stop := startRunCounting(t, p, dir, w, polled, func() { mu.Lock(); scans++; mu.Unlock() })
	defer stop()

	// While repair fails, it polls: the count climbs.
	n := count()
	eventually(t, "polling while degraded", func() bool { return count() >= n+3 })

	write(t, dir, "during.txt", "d")
	w.mu.Lock()
	w.repairErr = nil // the next Repair succeeds and clears degraded
	w.mu.Unlock()
	eventually(t, "the repair", func() bool { return !w.Degraded() })
	eventually(t, "during.txt found by the post-repair scan", func() bool { return manifestHas(s, "during.txt", sum([]byte("d"))) })

	time.Sleep(2 * polled.Poll) // let any scan already under way finish
	n = count()
	time.Sleep(8 * polled.Poll)
	if got := count(); got != n {
		t.Fatalf("%d more scans in eight polling periods after the repair; polling should have stopped", got-n)
	}
}

// A watcher whose reader dies closes its channel. The loop must fall back
// to polling, not spin and not stop publishing.
func TestRunPollsAfterTheWatcherStops(t *testing.T) {
	s, dir, w := release.NewDirStore(t.TempDir()), t.TempDir(), newFake()
	p := open(t, s)
	stop := startRun(t, p, dir, w, polled)
	defer stop()
	close(w.ch)
	write(t, dir, "after.txt", "a")
	eventually(t, "after.txt found by polling", func() bool { return manifestHas(s, "after.txt", sum([]byte("a"))) })
}

func TestRunWithNoWatcherPolls(t *testing.T) {
	s, dir := release.NewDirStore(t.TempDir()), t.TempDir()
	p := open(t, s)
	stop := startRun(t, p, dir, nil, polled)
	defer stop()
	write(t, dir, "x", "x")
	eventually(t, "x found by polling", func() bool { return manifestHas(s, "x", sum([]byte("x"))) })
}

func TestRunStopsWhenSuperseded(t *testing.T) {
	s, dir, w := release.NewDirStore(t.TempDir()), t.TempDir(), newFake()
	p, other := open(t, s), open(t, s)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	done := make(chan error, 1)
	go func() { done <- run(ctx, p, dir, eventsOnly, w) }()
	stage(t, other, "theirs", "t")
	commit(t, other)
	write(t, dir, "mine", "m")
	w.ch <- event{rel: "mine"}
	select {
	case err := <-done:
		if !errors.Is(err, ErrSuperseded) {
			t.Fatalf("err = %v, want ErrSuperseded", err)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("run did not stop after another writer published")
	}
}

func TestRunIgnoresAnEventForAPathThatEscapes(t *testing.T) {
	s, dir, w := release.NewDirStore(t.TempDir()), t.TempDir(), newFake()
	p := open(t, s)
	stop := startRun(t, p, dir, w, eventsOnly)
	w.ch <- event{rel: "../outside"}
	write(t, dir, "ok", "ok")
	w.ch <- event{rel: "ok"}
	eventually(t, "ok published", func() bool { return manifestHas(s, "ok", sum([]byte("ok"))) })
	if err := stop(); !errors.Is(err, context.Canceled) {
		t.Fatalf("run ended with %v; a bad path must be skipped, not fatal", err)
	}
}

// logCapture is a log.SetOutput target that is safe to write from the loop.
type logCapture struct {
	mu sync.Mutex
	b  strings.Builder
}

func (l *logCapture) Write(p []byte) (int, error) {
	l.mu.Lock()
	defer l.mu.Unlock()
	return l.b.Write(p)
}
func (l *logCapture) count(sub string) int {
	l.mu.Lock()
	defer l.mu.Unlock()
	return strings.Count(l.b.String(), sub)
}

func captureLog(t *testing.T) *logCapture {
	t.Helper()
	c := &logCapture{}
	log.SetOutput(c)
	t.Cleanup(func() { log.SetOutput(os.Stderr) })
	return c
}

// A repair is a tree walk: a cause that lasts must cost one attempt and
// one log line per polling period at most, not one per tick.
func TestRunRepairsAtMostOncePerPollAndLogsTheTransitionOnce(t *testing.T) {
	logs := captureLog(t)
	s, dir, w := release.NewDirStore(t.TempDir()), t.TempDir(), newFake()
	w.setDegraded(true)
	w.repairErr = errors.New("watch limit reached")
	p := open(t, s)
	stop := startRun(t, p, dir, w, polled)
	time.Sleep(20 * polled.Poll)
	stop()
	w.mu.Lock()
	n := w.repairs
	w.mu.Unlock()
	// Twenty periods; a tick is a quarter of the deadline, so per-tick
	// would be ~80. The bounds are wide so a loaded CI runner cannot flake.
	if n < 2 || n > 30 {
		t.Fatalf("%d repair attempts in twenty polling periods; want about twenty", n)
	}
	if got := logs.count("watch coverage is missing"); got != 1 {
		t.Fatalf("coverage-missing logged %d times, want once", got)
	}
	if got := logs.count("still missing"); got != 0 {
		t.Fatalf("a per-attempt line was logged %d times", got)
	}
}

func TestRunLogsWhenCoverageIsRestored(t *testing.T) {
	logs := captureLog(t)
	s, dir, w := release.NewDirStore(t.TempDir()), t.TempDir(), newFake()
	w.setDegraded(true)
	p := open(t, s) // Repair succeeds on its first attempt
	stop := startRun(t, p, dir, w, polled)
	eventually(t, "restored line", func() bool { return logs.count("watch coverage restored") == 1 })
	stop()
	if got := logs.count("watch coverage is missing"); got != 1 {
		t.Fatalf("missing logged %d times, want once", got)
	}
}

// failingPuts refuses every object upload, so no scan of a directory with a
// file in it can finish.
type failingPuts struct {
	release.Store
	mu   sync.Mutex
	puts int
}

func (f *failingPuts) Put(ctx context.Context, key string, body []byte, etag string) (string, error) {
	if strings.HasSuffix(key, "manifest.json") {
		return f.Store.Put(ctx, key, body, etag)
	}
	f.mu.Lock()
	f.puts++
	f.mu.Unlock()
	return "", errors.New("store unavailable")
}

func TestRunRetriesAFailingScanOncePerPollAndLogsItOnce(t *testing.T) {
	logs := captureLog(t)
	fs := &failingPuts{Store: release.NewDirStore(t.TempDir())}
	dir := t.TempDir()
	write(t, dir, "f", "v")
	p := open(t, fs)
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	go func() { done <- run(ctx, p, dir, polled, newFake()) }()
	time.Sleep(20 * polled.Poll)
	cancel()
	<-done
	fs.mu.Lock()
	n := fs.puts
	fs.mu.Unlock()
	if n < 2 || n > 30 { // per-tick would be ~80; see the repair test
		t.Fatalf("%d scan attempts in twenty polling periods; want about twenty", n)
	}
	if got := logs.count("scan failing"); got != 1 {
		t.Fatalf("the failure was logged %d times, want once", got)
	}
}

func TestRunWithZeroOptionsUsesTheDefaultsAndPublishesOnItsFirstScan(t *testing.T) {
	s, dir := release.NewDirStore(t.TempDir()), t.TempDir()
	write(t, dir, "a", "a")
	p := open(t, s)
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	go func() { done <- run(ctx, p, dir, Options{}, nil) }()
	eventually(t, "a published", func() bool { return manifestHas(s, "a", sum([]byte("a"))) })
	cancel()
	if err := <-done; !errors.Is(err, context.Canceled) {
		t.Fatal(err)
	}
}