package streamregistry

import (
	"context"
	"errors"
	"sync"
	"sync/atomic"
	"testing"
	"time"
)

// entryCommitter is the P3 handshake-commit surface frozen by the design
// gate. Keeping the assertion structural lets this tests-only red phase
// compile against the P2 baseline, where Entry intentionally does not yet
// implement TryCommit.
type entryCommitter interface {
	TryCommit() (owned bool, firstCause error)
}

func requireEntryCommitter(t *testing.T, entry *Entry) entryCommitter {
	t.Helper()
	committer, ok := any(entry).(entryCommitter)
	if !ok {
		t.Fatalf("Entry does not implement TryCommit() (bool, error)")
	}
	return committer
}

func TestEntryTryCommitRejectsCancelAfterStateWinsBeforeCauseApplied(t *testing.T) {
	cases := []struct {
		name  string
		cause error
		fire  func(*Registry, *Entry, context.CancelFunc)
	}{
		{
			name:  "generation",
			cause: CancelError{Reason: ReasonGenerationChange},
			fire: func(_ *Registry, entry *Entry, _ context.CancelFunc) {
				entry.Cancel(CancelError{Reason: ReasonGenerationChange})
			},
		},
		{
			name:  "parent",
			cause: CancelError{Reason: ReasonClientDisconnect},
			fire: func(_ *Registry, _ *Entry, cancelParent context.CancelFunc) {
				cancelParent()
			},
		},
		{
			name:  "deadline",
			cause: context.DeadlineExceeded,
			fire: func(_ *Registry, entry *Entry, _ context.CancelFunc) {
				entry.Cancel(context.DeadlineExceeded)
			},
		},
		{
			name:  "cancel-all",
			cause: ReasonDaemonShutdown,
			fire: func(registry *Registry, _ *Entry, _ context.CancelFunc) {
				registry.CancelAll()
			},
		},
	}

	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			parent, cancelParent := context.WithCancel(context.Background())
			defer cancelParent()
			registry := New()
			entry, err := registry.Register(parent, "session-a", 1, "server-a")
			if err != nil {
				t.Fatalf("Register: %v", err)
			}
			committer := requireEntryCommitter(t, entry)

			originalCancel := entry.cancel
			cancelEntered := make(chan error, 1)
			releaseCancel := make(chan struct{})
			entry.cancel = func(cause error) {
				cancelEntered <- cause
				<-releaseCancel
				originalCancel(cause)
			}

			fireDone := make(chan struct{})
			go func() {
				defer close(fireDone)
				tc.fire(registry, entry, cancelParent)
			}()

			var applied error
			select {
			case applied = <-cancelEntered:
			case <-time.After(time.Second):
				t.Fatal("Cancel did not reach the cancelCause barrier")
			}
			if !errors.Is(applied, tc.cause) {
				t.Fatalf("cancelCause argument = %v, want %v", applied, tc.cause)
			}

			owned, firstCause := committer.TryCommit()
			if owned {
				t.Fatal("TryCommit succeeded after Cancel had already won Entry.stateMu")
			}
			if !errors.Is(firstCause, tc.cause) {
				t.Fatalf("TryCommit firstCause = %v, want actual Entry cause %v", firstCause, tc.cause)
			}

			close(releaseCancel)
			select {
			case <-fireDone:
			case <-time.After(time.Second):
				t.Fatal("Cancel source did not return after cancelCause was released")
			}
			entry.Release()
		})
	}
}

func TestEntryTryCommitFirstThenCancelKeepsCommittedAndStableCause(t *testing.T) {
	cases := []struct {
		name  string
		cause error
	}{
		{name: "generation", cause: CancelError{Reason: ReasonGenerationChange}},
		{name: "parent", cause: CancelError{Reason: ReasonClientDisconnect}},
		{name: "deadline", cause: context.DeadlineExceeded},
		{name: "cancel-all", cause: CancelError{Reason: ReasonDaemonShutdown}},
	}

	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			registry := New()
			entry, err := registry.Register(context.Background(), "session-a", 1, "server-a")
			if err != nil {
				t.Fatalf("Register: %v", err)
			}
			committer := requireEntryCommitter(t, entry)
			owned, firstCause := committer.TryCommit()
			if !owned || firstCause != nil {
				t.Fatalf("TryCommit = (%v, %v), want (true, nil)", owned, firstCause)
			}

			var cancelCalls atomic.Int32
			originalCancel := entry.cancel
			entry.cancel = func(cause error) {
				cancelCalls.Add(1)
				originalCancel(cause)
			}

			entry.Cancel(tc.cause)
			entry.Cancel(CancelError{Reason: ReasonDaemonShutdown})
			if got := cancelCalls.Load(); got != 1 {
				t.Fatalf("cancelCause calls = %d, want exactly 1", got)
			}
			if cause := context.Cause(entry.Context()); !errors.Is(cause, tc.cause) {
				t.Fatalf("context cause = %v, want stable first cause %v", cause, tc.cause)
			}

			owned, firstCause = committer.TryCommit()
			if owned {
				t.Fatal("repeated TryCommit succeeded")
			}
			if !errors.Is(firstCause, tc.cause) {
				t.Fatalf("repeated TryCommit firstCause = %v, want %v", firstCause, tc.cause)
			}
			entry.Release()
		})
	}
}

func TestEntryReleaseFirstRejectsCommitAndCancelWithoutCause(t *testing.T) {
	registry := New()
	entry, err := registry.Register(context.Background(), "session-a", 1, "server-a")
	if err != nil {
		t.Fatalf("Register: %v", err)
	}
	committer := requireEntryCommitter(t, entry)

	entry.Release()
	entry.Cancel(CancelError{Reason: ReasonDaemonShutdown})
	owned, firstCause := committer.TryCommit()
	if owned || firstCause != nil {
		t.Fatalf("TryCommit after Release = (%v, %v), want (false, nil)", owned, firstCause)
	}
	if cause := context.Cause(entry.Context()); cause != nil {
		t.Fatalf("cause after Release-first = %v, want nil", cause)
	}
}

func TestEntryReleaseWaitsForCommittedCancelAppliedWithoutHoldingStateLock(t *testing.T) {
	registry := New()
	entry, err := registry.Register(context.Background(), "session-a", 1, "server-a")
	if err != nil {
		t.Fatalf("Register: %v", err)
	}
	committer := requireEntryCommitter(t, entry)
	owned, _ := committer.TryCommit()
	if !owned {
		t.Fatal("TryCommit did not acquire commit ownership")
	}

	originalCancel := entry.cancel
	cancelEntered := make(chan struct{})
	releaseCancel := make(chan struct{})
	entry.cancel = func(cause error) {
		close(cancelEntered)
		<-releaseCancel
		originalCancel(cause)
	}

	cancelDone := make(chan struct{})
	go func() {
		defer close(cancelDone)
		entry.Cancel(CancelError{Reason: ReasonDaemonShutdown})
	}()
	<-cancelEntered

	releaseDone := make(chan struct{})
	go func() {
		defer close(releaseDone)
		entry.Release()
	}()
	select {
	case <-releaseDone:
		t.Fatal("Release returned before cancelApplied")
	case <-time.After(20 * time.Millisecond):
	}

	// A repeated Cancel must be able to acquire stateMu while Release waits.
	// If Release waits on cancelApplied with stateMu held, this call deadlocks.
	repeatDone := make(chan struct{})
	go func() {
		defer close(repeatDone)
		entry.Cancel(CancelError{Reason: ReasonGenerationChange})
	}()
	select {
	case <-repeatDone:
	case <-time.After(time.Second):
		t.Fatal("repeated Cancel blocked: Release appears to hold stateMu while waiting")
	}

	close(releaseCancel)
	select {
	case <-cancelDone:
	case <-time.After(time.Second):
		t.Fatal("winning Cancel did not finish")
	}
	select {
	case <-releaseDone:
	case <-time.After(time.Second):
		t.Fatal("Release did not finish after cancelApplied")
	}
}

func TestEntryParentCancelAllReleaseThreeWayHasNoDeadlock(t *testing.T) {
	for iteration := 0; iteration < 100; iteration++ {
		parent, cancelParent := context.WithCancel(context.Background())
		registry := New()
		entry, err := registry.Register(parent, "session-a", uint64(iteration+1), "server-a")
		if err != nil {
			t.Fatalf("iteration %d Register: %v", iteration, err)
		}

		start := make(chan struct{})
		var wg sync.WaitGroup
		wg.Add(3)
		go func() {
			defer wg.Done()
			<-start
			cancelParent()
		}()
		go func() {
			defer wg.Done()
			<-start
			registry.CancelAll()
		}()
		go func() {
			defer wg.Done()
			<-start
			entry.Release()
		}()
		close(start)

		done := make(chan struct{})
		go func() {
			wg.Wait()
			close(done)
		}()
		select {
		case <-done:
		case <-time.After(time.Second):
			t.Fatalf("iteration %d deadlocked", iteration)
		}
	}
}

func TestEntryCancelNilNormalizesStableCause(t *testing.T) {
	registry := New()
	entry, err := registry.Register(context.Background(), "session-a", 1, "server-a")
	if err != nil {
		t.Fatalf("Register: %v", err)
	}
	committer := requireEntryCommitter(t, entry)

	entry.Cancel(nil)
	owned, firstCause := committer.TryCommit()
	if owned {
		t.Fatal("TryCommit succeeded after Cancel(nil)")
	}
	if !errors.Is(firstCause, context.Canceled) {
		t.Fatalf("TryCommit firstCause = %v, want context.Canceled", firstCause)
	}
	if cause := context.Cause(entry.Context()); !errors.Is(cause, context.Canceled) {
		t.Fatalf("context.Cause = %v, want context.Canceled", cause)
	}
	entry.Release()
}

// TestEntryTryCommitTwiceNeverReGrantsOwnership covers the codex5 blocker
// #4: a repeated TryCommit on an already-committed entry (with no
// intervening Cancel/Release) MUST NOT regress to (true, nil). Commit
// ownership is transferred exactly once; a second call returns (false,
// nil) so a programming error that double-calls TryCommit cannot act as
// if it had re-acquired the commit step. CommitObserved must remain
// closed from the first successful call (not re-closed, not blocked).
func TestEntryTryCommitTwiceNeverReGrantsOwnership(t *testing.T) {
	registry := New()
	entry, err := registry.Register(context.Background(), "session-a", 1, "server-a")
	if err != nil {
		t.Fatalf("Register: %v", err)
	}
	committer := requireEntryCommitter(t, entry)

	firstOwned, firstCause := committer.TryCommit()
	if !firstOwned || firstCause != nil {
		t.Fatalf("first TryCommit = (%v, %v), want (true, nil)", firstOwned, firstCause)
	}
	select {
	case <-entry.CommitObserved():
	default:
		t.Fatal("CommitObserved must be closed after the first successful TryCommit")
	}

	secondOwned, secondCause := committer.TryCommit()
	if secondOwned {
		t.Fatal("repeated TryCommit on entryCommitted re-granted ownership; want (false, nil)")
	}
	if secondCause != nil {
		t.Fatalf("repeated TryCommit firstCause = %v, want nil (no Cancel has fired)", secondCause)
	}
	// Observer seam stays closed (idempotent; no re-close panic, no stuck channel).
	select {
	case <-entry.CommitObserved():
	default:
		t.Fatal("CommitObserved must remain closed after a repeated TryCommit")
	}
	entry.Release()
}