package memorywatcher

import (
	"context"
	"errors"
	"sync"
	"testing"
	"time"

	"github.com/go-git/go-git/v5/plumbing/transport"
	"github.com/stretchr/testify/require"
)

type retryPusher struct {
	mu       sync.Mutex
	failures int
	err      error
	calls    []time.Time
}

type flakyCommitter struct {
	*fakeCommitter

	mu       sync.Mutex
	failures int
}

func (c *flakyCommitter) CommitPending(ctx context.Context, msg string, opts CommitOpts) (string, error) {
	c.mu.Lock()
	if c.failures > 0 {
		c.failures--
		c.mu.Unlock()
		return "", errors.New("commit unavailable")
	}
	c.mu.Unlock()
	return c.fakeCommitter.CommitPending(ctx, msg, opts)
}

func (c *flakyCommitter) setFailures(failures int) {
	c.mu.Lock()
	c.failures = failures
	c.mu.Unlock()
}

type blockingPusher struct {
	entered chan struct{}
	release chan struct{}
}

type blockingRetryPusher struct {
	mu      sync.Mutex
	calls   int
	entered chan struct{}
	release chan struct{}
}

func (p *blockingRetryPusher) Push(ctx context.Context, _ PushOpts) (string, error) {
	p.mu.Lock()
	p.calls++
	call := p.calls
	p.mu.Unlock()
	if call == 1 {
		return "", errors.New("upstream unavailable")
	}
	if call == 2 {
		close(p.entered)
		select {
		case <-p.release:
		case <-ctx.Done():
			return "", ctx.Err()
		}
	}
	return "pushed", nil
}

func (p *blockingPusher) Push(ctx context.Context, _ PushOpts) (string, error) {
	close(p.entered)
	select {
	case <-p.release:
		return "pushed", nil
	case <-ctx.Done():
		return "", ctx.Err()
	}
}

func (p *retryPusher) Push(context.Context, PushOpts) (string, error) {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.calls = append(p.calls, time.Now())
	if len(p.calls) <= p.failures {
		return "", p.err
	}
	return "pushed", nil
}

func (p *retryPusher) callCount() int {
	p.mu.Lock()
	defer p.mu.Unlock()
	return len(p.calls)
}

func flushOneCommit(t *testing.T, w *Watcher) {
	t.Helper()
	w.mu.Lock()
	w.hasPending = true
	w.mu.Unlock()
	w.flushCommit()
}

func TestWatcherRetriesFailedPushWithoutAnotherLocalWrite(t *testing.T) {
	pusher := &retryPusher{failures: 2, err: errors.New("upstream unavailable")}
	committer := &fakeCommitter{}
	w, err := New(Options{
		MemoryDir: t.TempDir(),
		Committer: committer,
		Resolve:   staticResolver(CommitContext{AgentID: "agt_retry"}),
		Push:      pusher,
		RetryMin:  10 * time.Millisecond,
		RetryMax:  20 * time.Millisecond,
	})
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, w.Close()) })

	flushOneCommit(t, w)
	status := w.Status()
	require.Equal(t, StatusPendingSync, status.State)
	require.Equal(t, StatusReasonPushFailed, status.Reason)
	require.False(t, status.RetryAt.IsZero())

	require.Eventually(t, func() bool {
		return pusher.callCount() == 3 && w.Status().State == StatusReady
	}, time.Second, 5*time.Millisecond)
	require.Equal(t, 1, committer.callCount(), "push retry must not create another local commit")
}

func TestWatcherReportsPendingWhilePushIsInFlight(t *testing.T) {
	pusher := &blockingPusher{entered: make(chan struct{}), release: make(chan struct{})}
	w, err := New(Options{
		MemoryDir: t.TempDir(),
		Committer: &fakeCommitter{},
		Resolve:   staticResolver(CommitContext{AgentID: "agt_retry"}),
		Push:      pusher,
	})
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, w.Close()) })

	w.mu.Lock()
	w.hasPending = true
	w.mu.Unlock()
	done := make(chan struct{})
	go func() {
		w.flushCommit()
		close(done)
	}()
	select {
	case <-pusher.entered:
	case <-time.After(time.Second):
		t.Fatal("push did not start")
	}
	require.Equal(t, Status{State: StatusPendingSync}, w.Status())
	close(pusher.release)
	select {
	case <-done:
	case <-time.After(time.Second):
		t.Fatal("flush did not finish")
	}
	require.Equal(t, StatusReady, w.Status().State)
}

func TestWatcherSerializesRetryPushWithLaterCommit(t *testing.T) {
	pusher := &blockingRetryPusher{entered: make(chan struct{}), release: make(chan struct{})}
	committer := &fakeCommitter{}
	w, err := New(Options{
		MemoryDir: t.TempDir(),
		Committer: committer,
		Resolve:   staticResolver(CommitContext{AgentID: "agt_retry"}),
		Push:      pusher,
		RetryMin:  10 * time.Millisecond,
		RetryMax:  10 * time.Millisecond,
	})
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, w.Close()) })

	flushOneCommit(t, w)
	select {
	case <-pusher.entered:
	case <-time.After(time.Second):
		t.Fatal("retry push did not start")
	}
	for range 100 {
		w.RetrySync()
	}
	w.mu.Lock()
	retryRunning := w.retryRunning
	retryTimer := w.retryTimer
	w.mu.Unlock()
	require.True(t, retryRunning)
	require.Nil(t, retryTimer, "explicit requests must coalesce onto the in-flight retry")
	require.True(t, w.Status().RetryAt.IsZero(), "in-flight retry must clear the scheduled retry time")

	w.mu.Lock()
	w.hasPending = true
	w.mu.Unlock()
	commitDone := make(chan struct{})
	go func() {
		w.flushCommit()
		close(commitDone)
	}()
	require.Never(t, func() bool { return committer.callCount() > 1 }, 50*time.Millisecond, 5*time.Millisecond,
		"a local commit must not race the in-flight retry push")
	close(pusher.release)
	select {
	case <-commitDone:
	case <-time.After(time.Second):
		t.Fatal("later commit did not resume after retry push")
	}
	require.Equal(t, 2, committer.callCount())
}

func TestWatcherStatusListenerCanRequestImmediateRetry(t *testing.T) {
	pusher := &retryPusher{failures: 1, err: errors.New("upstream unavailable")}
	var w *Watcher
	var retryOnce sync.Once
	var err error
	w, err = New(Options{
		MemoryDir: t.TempDir(),
		Committer: &fakeCommitter{},
		Resolve:   staticResolver(CommitContext{AgentID: "agt_retry"}),
		Push:      pusher,
		RetryMin:  time.Hour,
		RetryMax:  time.Hour,
		Status: func(status Status) {
			if status.Reason == StatusReasonPushFailed {
				retryOnce.Do(w.RetrySync)
			}
		},
	})
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, w.Close()) })

	flushOneCommit(t, w)
	require.Eventually(t, func() bool {
		return pusher.callCount() == 2 && w.Status().State == StatusReady
	}, time.Second, 5*time.Millisecond)
}

func TestWatcherExplicitRetryPreemptsBackoff(t *testing.T) {
	pusher := &retryPusher{failures: 1, err: errors.New("upstream unavailable")}
	w, err := New(Options{
		MemoryDir: t.TempDir(),
		Committer: &fakeCommitter{},
		Resolve:   staticResolver(CommitContext{AgentID: "agt_retry"}),
		Push:      pusher,
		RetryMin:  time.Hour,
		RetryMax:  time.Hour,
	})
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, w.Close()) })

	flushOneCommit(t, w)
	require.Equal(t, StatusPendingSync, w.Status().State)
	w.RetrySync()

	require.Eventually(t, func() bool {
		return pusher.callCount() == 2 && w.Status().State == StatusReady
	}, time.Second, 5*time.Millisecond)
}

func TestWatcherExplicitRetryRecoversFailedCommit(t *testing.T) {
	committer := &flakyCommitter{fakeCommitter: &fakeCommitter{}, failures: 1}
	pusher := &retryPusher{}
	w, err := New(Options{
		MemoryDir: t.TempDir(),
		Committer: committer,
		Resolve:   staticResolver(CommitContext{AgentID: "agt_retry"}),
		Push:      pusher,
	})
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, w.Close()) })

	flushOneCommit(t, w)
	require.Equal(t, Status{State: StatusDegraded, Reason: StatusReasonCommitFailed}, w.Status())
	require.Equal(t, 0, committer.callCount())

	w.RetrySync()
	require.Eventually(t, func() bool {
		return committer.callCount() == 1 && pusher.callCount() == 1 && w.Status().State == StatusReady
	}, time.Second, 5*time.Millisecond)
}

func TestWatcherReportsPendingDuringStartupPushProbe(t *testing.T) {
	pusher := &blockingPusher{entered: make(chan struct{}), release: make(chan struct{})}
	committer := &fakeCommitter{cleanWorktree: true}
	w, err := New(Options{
		MemoryDir:    t.TempDir(),
		Committer:    committer,
		Resolve:      staticResolver(CommitContext{AgentID: "agt_retry"}),
		Push:         pusher,
		RetryOnStart: true,
	})
	require.NoError(t, err)
	require.NoError(t, w.Start(t.Context()))
	t.Cleanup(func() { require.NoError(t, w.Close()) })

	select {
	case <-pusher.entered:
	case <-time.After(time.Second):
		t.Fatal("startup push probe did not start")
	}
	require.Equal(t, Status{State: StatusPendingSync}, w.Status())
	require.Equal(t, 1, committer.callCount(), "startup probe must inspect uncommitted local changes before pushing")
	close(pusher.release)
	require.Eventually(t, func() bool { return w.Status().State == StatusReady }, time.Second, 5*time.Millisecond)
}

func TestWatcherPushRetryCannotHideLaterCommitFailure(t *testing.T) {
	pusher := &retryPusher{failures: 1, err: errors.New("upstream unavailable")}
	committer := &flakyCommitter{fakeCommitter: &fakeCommitter{}}
	w, err := New(Options{
		MemoryDir: t.TempDir(),
		Committer: committer,
		Resolve:   staticResolver(CommitContext{AgentID: "agt_retry"}),
		Push:      pusher,
		RetryMin:  50 * time.Millisecond,
		RetryMax:  50 * time.Millisecond,
	})
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, w.Close()) })

	flushOneCommit(t, w)
	require.Equal(t, StatusPendingSync, w.Status().State)
	committer.setFailures(1)
	flushOneCommit(t, w)
	require.Equal(t, Status{State: StatusDegraded, Reason: StatusReasonCommitFailed}, w.Status())

	require.Eventually(t, func() bool { return pusher.callCount() == 2 }, time.Second, 5*time.Millisecond)
	require.Equal(t, Status{State: StatusDegraded, Reason: StatusReasonCommitFailed}, w.Status(),
		"pushing the older commit must not hide uncommitted local changes")

	w.RetrySync()
	require.Eventually(t, func() bool {
		return committer.callCount() == 2 && pusher.callCount() == 3 && w.Status().State == StatusReady
	}, time.Second, 5*time.Millisecond)
}

func TestWatcherClassifiesAuthenticationPushFailure(t *testing.T) {
	pusher := &retryPusher{failures: 1, err: transport.ErrAuthenticationRequired}
	w, err := New(Options{
		MemoryDir: t.TempDir(),
		Committer: &fakeCommitter{},
		Resolve:   staticResolver(CommitContext{AgentID: "agt_retry"}),
		Push:      pusher,
		RetryMin:  time.Hour,
		RetryMax:  time.Hour,
	})
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, w.Close()) })

	flushOneCommit(t, w)
	require.Equal(t, Status{
		State:   StatusPendingSync,
		Reason:  StatusReasonAuthFailed,
		RetryAt: w.Status().RetryAt,
	}, w.Status())
}

func TestWatcherRetryDelayIsExponentiallyBounded(t *testing.T) {
	w, err := New(Options{
		MemoryDir: t.TempDir(),
		Committer: &fakeCommitter{},
		Resolve:   staticResolver(CommitContext{}),
		RetryMin:  10 * time.Millisecond,
		RetryMax:  25 * time.Millisecond,
	})
	require.NoError(t, err)

	w.retryAttempt = 0
	require.Equal(t, 10*time.Millisecond, w.retryDelayLocked())
	w.retryAttempt = 1
	require.Equal(t, 20*time.Millisecond, w.retryDelayLocked())
	w.retryAttempt = 2
	require.Equal(t, 25*time.Millisecond, w.retryDelayLocked())
	w.retryAttempt = 20
	require.Equal(t, 25*time.Millisecond, w.retryDelayLocked())
}

func TestWatcherCloseCancelsScheduledPushRetry(t *testing.T) {
	pusher := &retryPusher{failures: 100, err: errors.New("upstream unavailable")}
	w, err := New(Options{
		MemoryDir: t.TempDir(),
		Committer: &fakeCommitter{},
		Resolve:   staticResolver(CommitContext{AgentID: "agt_retry"}),
		Push:      pusher,
		RetryMin:  50 * time.Millisecond,
		RetryMax:  50 * time.Millisecond,
	})
	require.NoError(t, err)

	flushOneCommit(t, w)
	require.Equal(t, 1, pusher.callCount())
	require.NoError(t, w.Close())
	time.Sleep(150 * time.Millisecond)
	require.Equal(t, 1, pusher.callCount(), "Close must prevent scheduled retry work")
}