package runtimeworker

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

	"github.com/nexuscrm/audit"
	"github.com/nexuscrm/persistence"
)

type auditOutboxControllerStub struct {
	prepareErr error
	runErr     error
	snapshot   audit.OutboxWorkerSnapshot

	prepareCalls atomic.Int64
	runCalls     atomic.Int64
	runStarted   chan struct{}
	startOnce    sync.Once
}

func (s *auditOutboxControllerStub) Prepare(context.Context) error {
	s.prepareCalls.Add(1)
	return s.prepareErr
}

func (s *auditOutboxControllerStub) Run(ctx context.Context) error {
	s.runCalls.Add(1)
	if s.runStarted != nil {
		s.startOnce.Do(func() { close(s.runStarted) })
	}
	if s.runErr != nil {
		return s.runErr
	}
	<-ctx.Done()
	return nil
}

func (s *auditOutboxControllerStub) Snapshot() audit.OutboxWorkerSnapshot { return s.snapshot }

type auditOutboxObservationResult struct {
	state audit.OutboxQueueState
	err   error
}

type auditOutboxObservationSourceStub struct {
	mu      sync.Mutex
	results []auditOutboxObservationResult
	calls   int
}

type blockingAuditOutboxObservationSource struct {
	started chan struct{}
	once    sync.Once
}

func (s *blockingAuditOutboxObservationSource) ObserveQueue(ctx context.Context) (audit.OutboxQueueState, error) {
	s.once.Do(func() { close(s.started) })
	<-ctx.Done()
	return audit.OutboxQueueState{}, ctx.Err()
}

func (s *auditOutboxObservationSourceStub) ObserveQueue(context.Context) (audit.OutboxQueueState, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.calls++
	if len(s.results) == 0 {
		return audit.OutboxQueueState{}, nil
	}
	result := s.results[0]
	if len(s.results) > 1 {
		s.results = s.results[1:]
	}
	return result.state, result.err
}

func testAuditOutboxOwnership() persistence.RuntimeWorkOwnership {
	return persistence.RuntimeWorkOwnership{
		WorkFamily:          audit.AuditHistoryFanoutWorkFamily,
		DeploymentID:        "audit-observation-test",
		WorkerReleaseDigest: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
		OwnershipEpoch:      4,
	}
}

func TestAuditOutboxRuntimeWorkerStartsStaleAndPreservesBusinessCounters(t *testing.T) {
	controller := &auditOutboxControllerStub{snapshot: audit.OutboxWorkerSnapshot{
		InFlight: 2, Attempts: 7, Failures: 1, Processed: 5,
	}}
	worker, err := newAuditOutboxRuntimeWorker(controller, func() (auditOutboxObservationSource, func() error, error) {
		return &auditOutboxObservationSourceStub{}, func() error { return nil }, nil
	})
	if err != nil {
		t.Fatal(err)
	}
	if err := worker.Prepare(t.Context()); err != nil {
		t.Fatal(err)
	}
	snapshot := worker.snapshot()
	if controller.prepareCalls.Load() != 1 || snapshot.InFlight != 2 || snapshot.Attempts != 7 || snapshot.Failures != 1 || snapshot.Processed != 5 {
		t.Fatalf("prepare calls=%d snapshot=%+v", controller.prepareCalls.Load(), snapshot)
	}
	if !snapshot.ObservationEnabled || !snapshot.ObservationStale || !snapshot.ObservationLastSuccess.IsZero() || snapshot.QueueDepth != 0 {
		t.Fatalf("initial observation snapshot=%+v, want unknown/stale", snapshot)
	}
}

func TestAuditOutboxRuntimeWorkerDoesNotObserveAtStartup(t *testing.T) {
	var opens atomic.Int64
	worker, err := newAuditOutboxRuntimeWorker(&auditOutboxControllerStub{}, func() (auditOutboxObservationSource, func() error, error) {
		opens.Add(1)
		return &auditOutboxObservationSourceStub{}, func() error { return nil }, nil
	})
	if err != nil {
		t.Fatal(err)
	}
	stop := worker.startObservationLoopWithTiming(t.Context(), time.Hour, time.Second)
	stop()
	if opens.Load() != 0 {
		t.Fatalf("startup observation opens=%d, want zero", opens.Load())
	}
}

func TestAuditOutboxRuntimeWorkerRetainsLastGoodObservationOnFailure(t *testing.T) {
	source := &auditOutboxObservationSourceStub{results: []auditOutboxObservationResult{
		{state: audit.OutboxQueueState{Depth: 7, PoisonDepth: 2, OldestAge: 3 * time.Second}},
		{err: errors.New("observation unavailable")},
	}}
	var releases atomic.Int64
	worker, err := newAuditOutboxRuntimeWorker(&auditOutboxControllerStub{}, func() (auditOutboxObservationSource, func() error, error) {
		return source, func() error { releases.Add(1); return nil }, nil
	})
	if err != nil {
		t.Fatal(err)
	}
	worker.refreshObservationWithTimeout(t.Context(), time.Second)
	good := worker.snapshot()
	if good.QueueDepth != 7 || good.PoisonDepth != 2 || good.OldestAge != 3*time.Second || good.ObservationStale || good.ObservationLastSuccess.IsZero() || good.OldestObservedAt.IsZero() {
		t.Fatalf("good snapshot=%+v", good)
	}
	worker.refreshObservationWithTimeout(t.Context(), time.Second)
	failed := worker.snapshot()
	if failed.QueueDepth != good.QueueDepth || failed.PoisonDepth != good.PoisonDepth || failed.OldestAge != good.OldestAge ||
		!failed.ObservationStale || failed.ObservationFailures != 1 || !failed.ObservationLastSuccess.Equal(good.ObservationLastSuccess) ||
		!failed.OldestObservedAt.Equal(good.OldestObservedAt) {
		t.Fatalf("failed snapshot=%+v, want last good %+v", failed, good)
	}
	if err := worker.closeObservation(); err != nil {
		t.Fatal(err)
	}
	if releases.Load() != 1 {
		t.Fatalf("observation releases=%d, want one", releases.Load())
	}
}

func TestAuditOutboxObservationFailureDoesNotStopBusinessController(t *testing.T) {
	controller := &auditOutboxControllerStub{runStarted: make(chan struct{})}
	worker, err := newAuditOutboxRuntimeWorker(controller, func() (auditOutboxObservationSource, func() error, error) {
		return nil, nil, errors.New("diagnostic pool unavailable")
	})
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithCancel(t.Context())
	done := make(chan error, 1)
	go func() {
		done <- worker.runWithObservationTiming(ctx, time.Millisecond, time.Second)
	}()
	select {
	case <-controller.runStarted:
	case <-time.After(time.Second):
		t.Fatal("business controller did not start")
	}
	deadline := time.Now().Add(time.Second)
	for worker.snapshot().ObservationFailures == 0 && time.Now().Before(deadline) {
		time.Sleep(time.Millisecond)
	}
	if worker.snapshot().ObservationFailures == 0 {
		t.Fatal("observation failure was not recorded")
	}
	select {
	case err := <-done:
		t.Fatalf("business controller stopped after observation failure: %v", err)
	default:
	}
	cancel()
	if err := <-done; err != nil {
		t.Fatal(err)
	}
}

func TestAuditOutboxObservationTimeoutIsSamplerOwned(t *testing.T) {
	source := &blockingAuditOutboxObservationSource{started: make(chan struct{})}
	worker, err := newAuditOutboxRuntimeWorker(&auditOutboxControllerStub{}, func() (auditOutboxObservationSource, func() error, error) {
		return source, func() error { return nil }, nil
	})
	if err != nil {
		t.Fatal(err)
	}
	started := time.Now()
	worker.refreshObservationWithTimeout(t.Context(), 10*time.Millisecond)
	select {
	case <-source.started:
	default:
		t.Fatal("blocking observation did not start")
	}
	if elapsed := time.Since(started); elapsed > time.Second {
		t.Fatalf("observation timeout took %s, want bounded sampler delay", elapsed)
	}
	snapshot := worker.snapshot()
	if snapshot.ObservationFailures != 1 || !snapshot.ObservationStale || snapshot.ObservationDuration < 10*time.Millisecond {
		t.Fatalf("timeout snapshot=%+v", snapshot)
	}
	if err := worker.closeObservation(); err != nil {
		t.Fatal(err)
	}
}

func TestAuditOutboxBusinessReturnStopsAndReleasesObservation(t *testing.T) {
	controller := &auditOutboxControllerStub{runErr: errors.New("business stopped")}
	var releases atomic.Int64
	worker, err := newAuditOutboxRuntimeWorker(controller, func() (auditOutboxObservationSource, func() error, error) {
		return &auditOutboxObservationSourceStub{}, func() error { releases.Add(1); return nil }, nil
	})
	if err != nil {
		t.Fatal(err)
	}
	// Open the lazy source first; Run must own its eventual release even when
	// the business controller returns an error immediately.
	worker.refreshObservationWithTimeout(t.Context(), time.Second)
	err = worker.runWithObservationTiming(t.Context(), time.Hour, time.Second)
	if !errors.Is(err, controller.runErr) {
		t.Fatalf("run error=%v, want %v", err, controller.runErr)
	}
	if releases.Load() != 1 {
		t.Fatalf("observation releases=%d, want one", releases.Load())
	}
}