package dispatcher

import (
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/desktop/internal/agent"
	"github.com/tidbcloud/AgentTeam/desktop/internal/agent/drivercontract"
	"github.com/tidbcloud/AgentTeam/desktop/internal/types"
)

func TestSessionResumeStateDirectorySyncFailurePropagatesAcrossOwnershipBoundary(t *testing.T) {
	workDir := t.TempDir()
	syncErr := errors.New("simulated directory fsync failure")
	previous := defaultSessionResumeStateStore
	defaultSessionResumeStateStore = &sessionResumeStateStore{
		syncDir: func(string) error { return syncErr },
	}
	t.Cleanup(func() { defaultSessionResumeStateStore = previous })

	_, err := newSessionDeliveryOwnershipStore(workDir).PutDeliveryOwnership(agent.DeliveryOwnershipRecord{
		Message:    types.InboxMessage{MessageID: "q-sync", Target: "#ops", Content: "accepted"},
		State:      agent.DeliveryOwnershipQueued,
		AcceptedAt: time.Now().UTC(),
	})
	require.ErrorIs(t, err, syncErr, "agent:deliver ACK must not succeed before the renamed directory entry is durable")

	// Rename happened before the injected fsync failure, so recovery may observe
	// the record. The important contract is that the accepting caller saw an
	// error and therefore could not acknowledge an unconfirmed durability point.
	records, loadErr := newSessionDeliveryOwnershipStore(workDir).LoadDeliveryOwnership()
	require.NoError(t, loadErr)
	require.Len(t, records, 1)
	require.Equal(t, "q-sync", records[0].MessageID())
}

func TestSessionResumeStateDeleteRetryResyncsAfterPriorRemoveSyncFailure(t *testing.T) {
	workDir := t.TempDir()
	require.NoError(t, os.WriteFile(sessionResumeStatePath(workDir), []byte(`{"lastSessionId":"thread-1"}`), 0o600))
	syncErr := errors.New("first directory sync failed")
	var calls int
	store := &sessionResumeStateStore{
		syncDir: func(string) error {
			calls++
			if calls == 1 {
				return syncErr
			}
			return nil
		},
	}

	require.ErrorIs(t, store.Delete(workDir), syncErr)
	_, err := os.Stat(sessionResumeStatePath(workDir))
	require.ErrorIs(t, err, os.ErrNotExist)
	require.NoError(t, store.Delete(workDir), "NotExist retry must still fsync the parent directory")
	require.Equal(t, 2, calls)
}

func TestSessionResumeStateStoreSerializesSameWorkDirUpdates(t *testing.T) {
	workDir := t.TempDir()
	const deliveries = 64

	var wg sync.WaitGroup
	errs := make(chan error, deliveries)
	acceptedAt := time.Now().UTC()
	for i := range deliveries {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			errs <- updateSessionResumeState(workDir, func(state *sessionResumeState) error {
				state.DeferredDeliveries = append(state.DeferredDeliveries, deferredDeliveryEntry{
					MessageID:  fmt.Sprintf("msg-%02d", i),
					Seq:        i + 1,
					AcceptedAt: acceptedAt.Add(time.Duration(i) * time.Millisecond),
				})
				return nil
			})
		}(i)
	}
	wg.Wait()
	close(errs)

	for err := range errs {
		require.NoError(t, err)
	}

	state, err := loadSessionResumeState(workDir)
	require.NoError(t, err)
	require.Len(t, state.DeferredDeliveries, deliveries)

	seen := make(map[string]struct{}, deliveries)
	for _, entry := range state.DeferredDeliveries {
		seen[entry.MessageID] = struct{}{}
	}
	for i := range deliveries {
		_, ok := seen[fmt.Sprintf("msg-%02d", i)]
		require.True(t, ok)
	}

	matches, err := filepath.Glob(filepath.Join(workDir, sessionResumeStateFilename+".tmp-*"))
	require.NoError(t, err)
	require.Empty(t, matches)
}

func TestSessionResumeStateStoreDoesNotBlockIndependentWorkDirs(t *testing.T) {
	blockedWorkDir := t.TempDir()
	independentWorkDir := t.TempDir()
	enteredBlockedUpdate := make(chan struct{})
	releaseBlockedUpdate := make(chan struct{})
	var releaseBlockedUpdateOnce sync.Once
	releaseBlocked := func() {
		releaseBlockedUpdateOnce.Do(func() {
			close(releaseBlockedUpdate)
		})
	}
	defer releaseBlocked()
	blockedUpdateDone := make(chan error, 1)

	go func() {
		blockedUpdateDone <- updateSessionResumeState(blockedWorkDir, func(state *sessionResumeState) error {
			close(enteredBlockedUpdate)
			<-releaseBlockedUpdate
			state.LastSessionID = "blocked"
			return nil
		})
	}()

	<-enteredBlockedUpdate

	independentUpdateDone := make(chan error, 1)
	go func() {
		independentUpdateDone <- updateSessionResumeState(independentWorkDir, func(state *sessionResumeState) error {
			state.LastSessionID = "independent"
			return nil
		})
	}()

	select {
	case err := <-independentUpdateDone:
		require.NoError(t, err)
	case <-time.After(250 * time.Millisecond):
		t.Fatal("independent workDir update was blocked by another workDir update")
	}

	releaseBlocked()
	require.NoError(t, <-blockedUpdateDone)

	state, err := loadSessionResumeState(blockedWorkDir)
	require.NoError(t, err)
	require.Equal(t, "blocked", state.LastSessionID)

	state, err = loadSessionResumeState(independentWorkDir)
	require.NoError(t, err)
	require.Equal(t, "independent", state.LastSessionID)
}

func TestSessionResumeStateTransactionalWritersPreserveIndependentFields(t *testing.T) {
	workDir := t.TempDir()
	policy := newRuntimePolicySnapshot("codex", "codex", types.AgentPermissionModeFullAccess, 1)
	epoch, err := persistStartConfigState(workDir, "codex", "http://server.test", "thread-old", &types.AgentStartConfig{
		Name:      "old-name",
		Runtime:   "codex",
		SessionID: "thread-old",
	}, policy)
	require.NoError(t, err)
	staleGuardState, err := loadSessionResumeState(workDir)
	require.NoError(t, err)

	entered := make(chan struct{})
	release := make(chan struct{})
	blockerDone := make(chan error, 1)
	go func() {
		blockerDone <- updateSessionResumeState(workDir, func(state *sessionResumeState) error {
			close(entered)
			<-release
			state.ActiveTaskTarget = types.ActiveTaskTarget{Channel: "ops", TaskNumber: 7}
			return nil
		})
	}()
	<-entered

	errCh := make(chan error, 5)
	go func() {
		_, err := newSessionDeliveryOwnershipStore(workDir).PutDeliveryOwnership(agent.DeliveryOwnershipRecord{
			Message:    types.InboxMessage{MessageID: "q1", Target: "#ops", Content: "accepted"},
			Input:      testRuntimeInputWithConfig(drivercontract.RuntimeInput{ID: "q1"}, "accepted"),
			State:      agent.DeliveryOwnershipQueued,
			AcceptedAt: time.Now().UTC(),
		})
		errCh <- err
	}()
	go func() {
		_, err := appendDeferredDelivery(workDir, 2, deferredRawMessage("deferred-1", "ops", "alice", "later", "2026-07-16T08:00:00Z"))
		errCh <- err
	}()
	go func() {
		_, err := applySessionResumeGuardToState(workDir, &staleGuardState, policy)
		errCh <- err
	}()
	go func() {
		_, _, _, err := persistAgentIdentityStartConfig(workDir, "agent-1", "new-name", "New Name")
		errCh <- err
	}()
	go func() {
		_, err := PersistResumeSafety(workDir, "http://server.test", "thread-new", epoch, 1, true)
		errCh <- err
	}()

	close(release)
	require.NoError(t, <-blockerDone)
	for range 5 {
		require.NoError(t, <-errCh)
	}

	state, err := loadSessionResumeState(workDir)
	require.NoError(t, err)
	// Whichever transaction acquired the lock first, the accepted live owner
	// must win over the Process-local safe snapshot. PersistResumeSafety
	// re-checks the journal in its transaction and PutDeliveryOwnership itself
	// fail-closes the same state when it runs second.
	require.False(t, state.SessionResumeSafe)
	require.Equal(t, uint64(1), state.SessionResumeSafetyVersion)
	require.Equal(t, "thread-new", state.LastSessionID)
	require.NotNil(t, state.LastStartConfig)
	require.Equal(t, "new-name", state.LastStartConfig.Name)
	require.Equal(t, "New Name", state.LastStartConfig.DisplayName)
	require.Equal(t, "thread-new", state.LastStartConfig.SessionID)
	require.Len(t, state.RuntimeDeliveryOwnership, 1)
	require.Equal(t, "q1", state.RuntimeDeliveryOwnership[0].MessageID())
	require.Len(t, state.DeferredDeliveries, 1)
	require.Equal(t, "deferred-1", state.DeferredDeliveries[0].MessageID)
	require.Equal(t, int64(7), state.ActiveTaskTarget.TaskNumber)
	savedPolicy, ok := savedRuntimePolicySnapshot(state)
	require.True(t, ok)
	require.Equal(t, policy.Fingerprint, savedPolicy.Fingerprint)
}