package tcp

import (
	"context"
	"testing"
	"time"

	"spark-server/internal/pkg/errs"
)

func TestTaskPoolRunsJob(t *testing.T) {
	p := NewTaskPool(1, 4)
	defer p.Close()
	done := make(chan struct{})
	err := p.Submit(context.Background(), func() { close(done) })
	if err != nil {
		t.Fatal(err)
	}
	select {
	case <-done:
	case <-time.After(time.Second):
		t.Fatal("timeout")
	}
}

func TestTaskPoolRecoversPanic(t *testing.T) {
	p := NewTaskPool(1, 4)
	defer p.Close()
	if err := p.Submit(context.Background(), func() { panic("boom") }); err != nil {
		t.Fatal(err)
	}
	done := make(chan struct{})
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		if err := p.Submit(context.Background(), func() { close(done) }); err == nil {
			break
		}
		time.Sleep(10 * time.Millisecond)
	}
	select {
	case <-done:
	case <-time.After(time.Second):
		t.Fatal("worker lost after panic")
	}
}

func TestTaskPoolBusy(t *testing.T) {
	p := NewTaskPool(1, 1)
	defer p.Close()
	started := make(chan struct{})
	block := make(chan struct{})
	if err := p.Submit(context.Background(), func() { close(started); <-block }); err != nil {
		t.Fatal(err)
	}
	<-started
	if err := p.Submit(context.Background(), func() {}); err != nil {
		t.Fatal(err)
	}
	err := p.Submit(context.Background(), func() {})
	close(block)
	if err != errs.ErrBusy {
		t.Fatalf("want busy, got %v", err)
	}
}

func TestTaskPoolClosed(t *testing.T) {
	p := NewTaskPool(1, 1)
	p.Close()
	if !p.Closed() {
		t.Fatal("closed")
	}
}