package task

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

func TestCapacityDrainPanicAndClose(t *testing.T) {
	r, err := New(Config{Workers: 1, QueueSize: 1, OnError: func(string, error) { panic("callback failure") }})
	if err != nil {
		t.Fatal(err)
	}
	defer r.Close()
	started, release := make(chan struct{}), make(chan struct{})
	first, _ := r.Submit(context.Background(), "first", func(context.Context) error { close(started); <-release; return nil })
	<-started
	second, _ := r.Submit(context.Background(), "panic", func(context.Context) error { panic("job failure") })
	if _, err := r.Submit(context.Background(), "overflow", func(context.Context) error { return nil }); !errors.Is(err, ErrQueueFull) {
		t.Fatal(err)
	}
	done := make(chan error, 1)
	go func() { done <- r.Close() }()
	close(release)
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	if err := first.Wait(ctx); err != nil {
		t.Fatal(err)
	}
	if err := second.Wait(ctx); err == nil {
		t.Fatal("panic lost")
	}
	if err := <-done; err != nil {
		t.Fatal(err)
	}
	if s := r.Stats(); s.Completed != 2 || s.Failed != 1 || s.Active != 0 || s.Queued != 0 {
		t.Fatal(s)
	}
	if _, err := r.Submit(ctx, "late", func(context.Context) error { return nil }); !errors.Is(err, ErrClosed) {
		t.Fatal(err)
	}
}

func TestShutdownDeadlineAndQueuedCancellation(t *testing.T) {
	r, _ := New(Config{Workers: 1, QueueSize: 1, ShutdownTimeout: 20 * time.Millisecond, OnError: func(string, error) {}})
	started, release := make(chan struct{}), make(chan struct{})
	var queued atomic.Bool
	first, _ := r.Submit(context.Background(), "ignoresCancel", func(context.Context) error { close(started); <-release; return nil })
	<-started
	second, _ := r.Submit(context.Background(), "queued", func(context.Context) error { queued.Store(true); return nil })
	if err := r.Close(); !errors.Is(err, context.DeadlineExceeded) {
		t.Fatal(err)
	}
	close(release)
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	if err := first.Wait(ctx); !errors.Is(err, context.Canceled) {
		t.Fatal(err)
	}
	if err := second.Wait(ctx); !errors.Is(err, context.Canceled) {
		t.Fatal(err)
	}
	if queued.Load() {
		t.Fatal("ran queued work after forced shutdown")
	}
	if err := r.Shutdown(ctx); err != nil {
		t.Fatal(err)
	}
}

func TestIntervalsDoNotOverlapAndTaskDeadline(t *testing.T) {
	r, _ := New(Config{Workers: 2, TaskTimeout: 25 * time.Millisecond, OnError: func(string, error) {}})
	defer r.Close()
	var active, maxActive atomic.Int32
	job := func(ctx context.Context) error {
		n := active.Add(1)
		if n > maxActive.Load() {
			maxActive.Store(n)
		}
		defer active.Add(-1)
		<-ctx.Done()
		return ctx.Err()
	}
	if err := r.Every("periodic", 3*time.Millisecond, job); err != nil {
		t.Fatal(err)
	}
	if err := r.Every("periodic", time.Second, job); err == nil {
		t.Fatal("duplicate schedule accepted")
	}
	deadline := time.Now().Add(time.Second)
	for r.Stats().Completed < 2 && time.Now().Before(deadline) {
		time.Sleep(2 * time.Millisecond)
	}
	if err := r.Close(); err != nil {
		t.Fatal(err)
	}
	if maxActive.Load() != 1 || r.Stats().Skipped == 0 || r.Stats().Failed < 2 {
		t.Fatal(maxActive.Load(), r.Stats())
	}
}