worker/compliance_test.go

package worker

import (
	"context"
	"errors"
	"go/ast"
	"go/parser"
	"go/token"
	"os"
	"path/filepath"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	"code.nochebuena.dev/einherjar/contracts/lifecycle"
	"code.nochebuena.dev/einherjar/contracts/logging"
)

// Compile-time interface checks (CT-5 / I-8).
var _ lifecycle.Component = (Component)(nil)
var _ Provider = (Component)(nil)

// --- CT-6: at most one exported TypeSpec per non-test, non-doc file ---

func TestAtMostOneExportedTypePerFile(t *testing.T) {
	fset := token.NewFileSet()
	pkgs, err := parser.ParseDir(fset, ".", func(fi os.FileInfo) bool {
		name := fi.Name()
		return !strings.HasSuffix(name, "_test.go") && name != "doc.go"
	}, 0)
	if err != nil {
		t.Fatalf("parse: %v", err)
	}
	for _, pkg := range pkgs {
		for path, file := range pkg.Files {
			base := filepath.Base(path)
			count := 0
			for _, decl := range file.Decls {
				gd, ok := decl.(*ast.GenDecl)
				if !ok {
					continue
				}
				for _, spec := range gd.Specs {
					ts, ok := spec.(*ast.TypeSpec)
					if ok && ts.Name.IsExported() {
						count++
					}
				}
			}
			if count > 1 {
				t.Errorf("%s: %d exported TypeSpecs (max 1)", base, count)
			}
		}
	}
}

// --- Config defaults (S-4) ---

func TestDefaultConfig_OptionalFields(t *testing.T) {
	cfg := DefaultConfig()
	if cfg.PoolSize == 0 {
		t.Error("PoolSize must have a default")
	}
	if cfg.BufferSize == 0 {
		t.Error("BufferSize must have a default")
	}
	if cfg.ShutdownTimeout == 0 {
		t.Error("ShutdownTimeout must have a default")
	}
	// TaskTimeout == 0 is intentional: means no per-task deadline
}

// --- New / lifecycle ---

func TestNew_NotNil(t *testing.T) {
	if New(newLogger(), Config{}) == nil {
		t.Fatal("New returned nil")
	}
}

func TestNew_AppliesDefaults(t *testing.T) {
	c := New(newLogger(), Config{PoolSize: 0, BufferSize: 0}).(*workerImpl)
	if c.cfg.PoolSize != 5 {
		t.Errorf("PoolSize = %d, want 5", c.cfg.PoolSize)
	}
	if c.cfg.BufferSize != 100 {
		t.Errorf("BufferSize = %d, want 100", c.cfg.BufferSize)
	}
}

// --- Behavioral tests ---

func TestWorker_DispatchAndExecute(t *testing.T) {
	done := make(chan struct{})
	c := startWorker(t, Config{PoolSize: 1, BufferSize: 10, ShutdownTimeout: time.Second})

	c.Dispatch(func(ctx context.Context) error {
		close(done)
		return nil
	})

	select {
	case <-done:
	case <-time.After(time.Second):
		t.Fatal("task not executed in time")
	}
	_ = c.OnStop()
}

func TestWorker_BackpressureFull(t *testing.T) {
	block := make(chan struct{})
	c := startWorker(t, Config{PoolSize: 1, BufferSize: 1, ShutdownTimeout: time.Second})

	c.Dispatch(func(ctx context.Context) error { <-block; return nil }) // fills worker
	c.Dispatch(func(ctx context.Context) error { return nil })          // fills buffer

	ok := c.Dispatch(func(ctx context.Context) error { return nil }) // should be dropped
	if ok {
		t.Error("expected Dispatch to return false when queue is full")
	}
	close(block)
	_ = c.OnStop()
}

func TestWorker_OnStop_DrainsQueue(t *testing.T) {
	var count int64
	c := startWorker(t, Config{PoolSize: 2, BufferSize: 50, ShutdownTimeout: 5 * time.Second})

	for i := 0; i < 10; i++ {
		c.Dispatch(func(ctx context.Context) error {
			atomic.AddInt64(&count, 1)
			return nil
		})
	}
	_ = c.OnStop()

	if atomic.LoadInt64(&count) != 10 {
		t.Errorf("expected 10 tasks completed, got %d", count)
	}
}

func TestWorker_OnStop_Timeout(t *testing.T) {
	c := startWorker(t, Config{PoolSize: 1, BufferSize: 1,
		ShutdownTimeout: 50 * time.Millisecond})

	c.Dispatch(func(ctx context.Context) error {
		time.Sleep(500 * time.Millisecond)
		return nil
	})

	start := time.Now()
	_ = c.OnStop()
	elapsed := time.Since(start)

	if elapsed > 300*time.Millisecond {
		t.Errorf("OnStop blocked too long: %v", elapsed)
	}
}

func TestWorker_TaskTimeout(t *testing.T) {
	var ctxCancelled int64
	c := startWorker(t, Config{
		PoolSize: 1, BufferSize: 10,
		TaskTimeout:     50 * time.Millisecond,
		ShutdownTimeout: time.Second,
	})

	done := make(chan struct{})
	c.Dispatch(func(ctx context.Context) error {
		defer close(done)
		select {
		case <-ctx.Done():
			atomic.StoreInt64(&ctxCancelled, 1)
		case <-time.After(500 * time.Millisecond):
		}
		return nil
	})

	select {
	case <-done:
	case <-time.After(time.Second):
		t.Fatal("task did not complete in time")
	}
	if atomic.LoadInt64(&ctxCancelled) != 1 {
		t.Error("expected task context to be cancelled by TaskTimeout")
	}
	_ = c.OnStop()
}

func TestWorker_MultipleWorkers(t *testing.T) {
	const n = 5
	started := make(chan struct{}, n)
	block := make(chan struct{})

	c := startWorker(t, Config{PoolSize: n, BufferSize: n, ShutdownTimeout: time.Second})

	for i := 0; i < n; i++ {
		c.Dispatch(func(ctx context.Context) error {
			started <- struct{}{}
			<-block
			return nil
		})
	}

	timer := time.NewTimer(time.Second)
	defer timer.Stop()
	for i := 0; i < n; i++ {
		select {
		case <-started:
		case <-timer.C:
			t.Fatalf("only %d/%d workers started concurrently", i, n)
		}
	}
	close(block)
	_ = c.OnStop()
}

func TestWorker_TaskError(t *testing.T) {
	c := startWorker(t, Config{PoolSize: 1, BufferSize: 10, ShutdownTimeout: time.Second})

	done := make(chan struct{})
	c.Dispatch(func(ctx context.Context) error {
		defer close(done)
		return errors.New("task error")
	})

	select {
	case <-done:
	case <-time.After(time.Second):
		t.Fatal("task did not run")
	}
	_ = c.OnStop()
}

func TestWorker_Len(t *testing.T) {
	c := New(newLogger(), Config{PoolSize: 1, BufferSize: 10, ShutdownTimeout: time.Second})
	if err := c.OnInit(); err != nil {
		t.Fatalf("OnInit: %v", err)
	}
	// Workers are not running yet — queued tasks stay in the channel.
	blocked := make(chan struct{})
	c.Dispatch(func(ctx context.Context) error { <-blocked; return nil })
	c.Dispatch(func(ctx context.Context) error { <-blocked; return nil })
	if got := c.Len(); got != 2 {
		t.Errorf("Len() = %d, want 2", got)
	}
	close(blocked)
	_ = c.OnStop()
}

func TestWorker_Lifecycle(t *testing.T) {
	c := New(newLogger(), Config{PoolSize: 2, BufferSize: 10, ShutdownTimeout: time.Second})
	if err := c.OnInit(); err != nil {
		t.Fatalf("OnInit: %v", err)
	}
	if err := c.OnStart(); err != nil {
		t.Fatalf("OnStart: %v", err)
	}
	done := make(chan struct{})
	c.Dispatch(func(ctx context.Context) error { close(done); return nil })
	select {
	case <-done:
	case <-time.After(time.Second):
		t.Fatal("task not executed")
	}
	if err := c.OnStop(); err != nil {
		t.Fatalf("OnStop: %v", err)
	}
}

// --- helpers ---

type stubLogger struct{}

func newLogger() *stubLogger { return &stubLogger{} }

func (s *stubLogger) Debug(msg string, args ...any)                  {}
func (s *stubLogger) Info(msg string, args ...any)                   {}
func (s *stubLogger) Warn(msg string, args ...any)                   {}
func (s *stubLogger) Error(msg string, err error, args ...any)       {}
func (s *stubLogger) With(args ...any) logging.Logger                { return s }
func (s *stubLogger) WithContext(ctx context.Context) logging.Logger { return s }

func startWorker(t *testing.T, cfg Config) Component {
	t.Helper()
	c := New(newLogger(), cfg)
	if err := c.OnInit(); err != nil {
		t.Fatalf("OnInit: %v", err)
	}
	if err := c.OnStart(); err != nil {
		t.Fatalf("OnStart: %v", err)
	}
	return c
}