package engine

import (
	"context"
	"fmt"
	"sync"
	"time"
)

type Config struct {
	Mode      string `json:"mode"`
	Messages  int    `json:"messages"`
	Workers   int    `json:"workers"`
	DelayMS   int    `json:"delayMs"`
	CancelMS  int    `json:"cancelMs"`
	FailEvery int    `json:"failEvery"`
}
type Result struct {
	Mode            string   `json:"mode"`
	Started         int      `json:"started"`
	Completed       int      `json:"completed"`
	Failed          int      `json:"failed"`
	PeakActive      int      `json:"peakActive"`
	BlockedAtCancel int      `json:"blockedAtCancel"`
	FinalActive     int      `json:"finalActive"`
	Canceled        bool     `json:"canceled"`
	ElapsedMS       int64    `json:"elapsedMs"`
	Trace           []string `json:"trace"`
}

func (c Config) Validate() error {
	if c.Mode != "fixed" && c.Mode != "broken" {
		return fmt.Errorf("choose fixed or broken")
	}
	if c.Messages < 1 || c.Messages > 64 || c.Workers < 1 || c.Workers > 8 {
		return fmt.Errorf("use 1–64 messages and 1–8 workers")
	}
	if c.DelayMS < 10 || c.DelayMS > 1000 || c.CancelMS < 0 || c.CancelMS > 2000 || c.FailEvery < 0 || c.FailEvery > 64 {
		return fmt.Errorf("delay, cancellation, or error frequency is out of range")
	}
	return nil
}

type counters struct {
	mu     sync.Mutex
	result Result
	active int
}

func (c *counters) begin() {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.active++
	c.result.Started++
	if c.active > c.result.PeakActive {
		c.result.PeakActive = c.active
	}
}
func (c *counters) end(outcome string) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.active--
	if outcome == "completed" {
		c.result.Completed++
	}
	if outcome == "failed" {
		c.result.Failed++
	}
}
func (c *counters) note(text string) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.result.Trace = append(c.result.Trace, text)
}
func (c *counters) cancelSnapshot() {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.result.Canceled = true
	c.result.BlockedAtCancel = c.active
}
func Run(parent context.Context, config Config) (Result, error) {
	if err := config.Validate(); err != nil {
		return Result{}, err
	}
	start := time.Now()
	ctx, cancel := context.WithCancel(parent)
	defer cancel()
	var cancelTimer *time.Timer
	if config.CancelMS > 0 {
		cancelTimer = time.AfterFunc(time.Duration(config.CancelMS)*time.Millisecond, cancel)
		defer cancelTimer.Stop()
	}
	c := &counters{result: Result{Mode: config.Mode, Trace: []string{}}}
	if config.Mode == "broken" {
		runBroken(ctx, config, c)
	} else {
		runFixed(ctx, config, c)
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	c.result.FinalActive = c.active
	c.result.ElapsedMS = time.Since(start).Milliseconds()
	return c.result, nil
}
func runFixed(ctx context.Context, config Config, c *counters) {
	jobs := make(chan int, config.Workers)
	var workers sync.WaitGroup
	c.note(fmt.Sprintf("A bounded queue feeds %d workers.", config.Workers))
	for range config.Workers {
		workers.Go(func() {
			for {
				select {
				case <-ctx.Done():
					return
				case id, open := <-jobs:
					if !open {
						return
					}
					if ctx.Err() != nil {
						return
					}
					c.begin()
					timer := time.NewTimer(time.Duration(config.DelayMS) * time.Millisecond)
					select {
					case <-ctx.Done():
						timer.Stop()
						c.end("canceled")
						return
					case <-timer.C:
						if ctx.Err() != nil {
							c.end("canceled")
							return
						}
						if config.FailEvery > 0 && id%config.FailEvery == 0 {
							c.end("failed")
						} else {
							c.end("completed")
						}
					}
				}
			}
		})
	}
	// This goroutine is the sole owner of closing jobs.
	var producer sync.WaitGroup
	producer.Go(func() {
		defer close(jobs)
		for id := 1; id <= config.Messages; id++ {
			select {
			case <-ctx.Done():
				return
			case jobs <- id:
			}
		}
	})
	producer.Wait()
	workers.Wait()
	if ctx.Err() != nil {
		c.cancelSnapshot()
		c.note("Cancellation reached producers and workers. All workers joined before returning.")
	} else {
		c.note("The producer closed its queue. Workers drained it and returned.")
	}
}
func runBroken(ctx context.Context, config Config, c *counters) {
	// Deliberately broken ownership: canceled collectors leave unbuffered sends blocked.
	// The teaching harness releases the blockage after observing it; it never retains leaks.
	results := make(chan int)
	release := make(chan struct{})
	ready := make(chan struct{}, config.Messages)
	var workers sync.WaitGroup
	c.note("One goroutine starts per message; the configured worker limit is ignored.")
	for id := 1; id <= config.Messages; id++ {
		workers.Go(func() {
			c.begin()
			ready <- struct{}{}
			timer := time.NewTimer(time.Duration(config.DelayMS) * time.Millisecond)
			defer timer.Stop()
			select {
			case <-release:
				c.end("canceled")
				return
			case <-timer.C:
			}
			select {
			case <-release:
				c.end("canceled")
			case results <- id:
				if config.FailEvery > 0 && id%config.FailEvery == 0 {
					c.end("failed")
				} else {
					c.end("completed")
				}
			}
		})
	}
	for range config.Messages {
		<-ready
	}
	for received := 0; received < config.Messages; received++ {
		select {
		case <-ctx.Done():
			c.cancelSnapshot()
			c.note("The collector stopped. Workers do not observe its context and can block on result delivery.")
			close(release)
			workers.Wait()
			c.note("The teaching harness released those workers after recording the failure.")
			return
		case <-results:
		}
	}
	close(release)
	workers.Wait()
	c.note("All results arrived, but peak work was not bounded by the worker setting.")
}
