package engine

import (
	"context"
	"log/slog"
	"sync"
	"testing"
	"time"

	"tangled.org/core/spindle/config"
	"tangled.org/core/spindle/models"
	"tangled.org/core/spindle/observability"
)

type queueRun struct {
	statuses map[string]models.StatusKind
	finished []slog.Record
	metrics  *observability.Metrics
}

// runQueued starts workflows a and b behind a single slot
func runQueued(t *testing.T, timeout time.Duration, runStep func(ctx context.Context, wid models.WorkflowId) error, during func(holder, waiter models.WorkflowId)) queueRun {
	t.Helper()
	var records []slog.Record
	logger := slog.New(&captureHandler{records: &records, mu: &sync.Mutex{}})
	metrics := observability.NewMetrics()
	testDB := newTestDB(t)
	eng := &slottedEngine{mockEngine: &mockEngine{timeout: timeout}, slots: NewSemaphoreSlotter(1)}
	first := make(chan models.WorkflowId, 1)
	var once sync.Once
	eng.runStepFunc = func(ctx context.Context, wid models.WorkflowId, _ int, _ models.WorkflowLogger) error {
		once.Do(func() { first <- wid })
		return runStep(ctx, wid)
	}

	pipelineId := models.PipelineId("queue-timeout")
	step := []models.Step{mockStep{name: "step"}}
	pipeline := &models.Pipeline{Workflows: map[models.Engine][]models.Workflow{
		eng: {{Name: "a", Steps: step}, {Name: "b", Steps: step}},
	}}
	cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}}

	var wg sync.WaitGroup
	wg.Go(func() {
		StartWorkflows(logger, nil, cfg, nil, nil, testDB, nil, nil, nil, observability.WithMetrics(context.Background(), metrics), pipeline, pipelineId)
	})
	if during != nil {
		holder := <-first
		waiter := models.WorkflowId{PipelineId: pipelineId, Name: "a"}
		if holder.Name == "a" {
			waiter.Name = "b"
		}
		during(holder, waiter)
	}
	wg.Wait()

	run := queueRun{statuses: map[string]models.StatusKind{}, metrics: metrics}
	for _, name := range []string{"a", "b"} {
		st, err := testDB.GetStatus(models.WorkflowId{PipelineId: pipelineId, Name: name})
		if err != nil {
			t.Fatalf("GetStatus(%s) error = %v", name, err)
		}
		run.statuses[name] = st
	}
	for _, r := range records {
		if r.Message == "workflow finished" {
			run.finished = append(run.finished, r)
		}
	}
	return run
}

func TestQueueWaitDoesNotCountAgainstTheWorkflowTimeout(t *testing.T) {
	// each run takes 250ms. the second one finishes 500ms after both were queued, past the 400ms timeout
	statuses := runQueued(t, 400*time.Millisecond, func(ctx context.Context, _ models.WorkflowId) error {
		select {
		case <-time.After(250 * time.Millisecond):
			return nil
		case <-ctx.Done():
			return ctx.Err()
		}
	}, nil).statuses
	for name, st := range statuses {
		if st != models.StatusKindSuccess {
			t.Errorf("workflow %s status = %s, want success", name, st)
		}
	}
}

func TestWorkflowThatRunsPastItsTimeoutAfterQueueingStillTimesOut(t *testing.T) {
	timeout := 400 * time.Millisecond
	var mu sync.Mutex
	var ran time.Duration
	holderSeen := false
	statuses := runQueued(t, timeout, func(ctx context.Context, _ models.WorkflowId) error {
		mu.Lock()
		holder := !holderSeen
		holderSeen = true
		mu.Unlock()
		if holder {
			time.Sleep(250 * time.Millisecond)
			return nil
		}
		start := time.Now()
		<-ctx.Done()
		mu.Lock()
		ran = time.Since(start)
		mu.Unlock()
		return ctx.Err()
	}, nil).statuses

	timedOut := 0
	for _, st := range statuses {
		if st == models.StatusKindTimeout {
			timedOut++
		}
	}
	if timedOut != 1 {
		t.Errorf("statuses = %v, want exactly one timeout", statuses)
	}
	if ran < timeout-50*time.Millisecond {
		t.Errorf("queued workflow ran for %s before timing out, want about %s", ran, timeout)
	}
}

func TestCancelInterruptsAQueuedWorkflow(t *testing.T) {
	release := make(chan struct{})
	var mu sync.Mutex
	var ranSteps []string
	statuses := runQueued(t, 5*time.Second, func(ctx context.Context, wid models.WorkflowId) error {
		mu.Lock()
		ranSteps = append(ranSteps, wid.Name)
		mu.Unlock()
		select {
		case <-release:
			return nil
		case <-ctx.Done():
			return ctx.Err()
		}
	}, func(_, waiter models.WorkflowId) {
		// nothing reports when b starts waiting for the slot. this sleep gives it time to get there
		time.Sleep(300 * time.Millisecond)
		CancelWorkflow(waiter)
		time.Sleep(100 * time.Millisecond)
		close(release)
	}).statuses

	if len(ranSteps) != 1 {
		t.Errorf("steps ran for %v, want only the slot holder", ranSteps)
	}
	cancelled := 0
	for _, st := range statuses {
		if st == models.StatusKindCancelled {
			cancelled++
		}
	}
	if cancelled != 1 {
		t.Errorf("statuses = %v, want exactly one cancelled", statuses)
	}
}

func TestQueueWaitIsLeftOutOfTheWorkflowDuration(t *testing.T) {
	// each run takes 250ms and the second one waits about 250ms for the slot first
	run := runQueued(t, 5*time.Second, func(context.Context, models.WorkflowId) error {
		time.Sleep(250 * time.Millisecond)
		return nil
	}, nil)

	if len(run.finished) != 2 {
		t.Fatalf("got %d workflow finished logs, want 2", len(run.finished))
	}
	for _, r := range run.finished {
		r.Attrs(func(a slog.Attr) bool {
			if a.Key == "duration_seconds" && a.Value.Float64() > 0.4 {
				t.Errorf("logged duration %vs, want about 0.25s of running", a.Value.Float64())
			}
			return true
		})
	}

	families, err := run.metrics.Registry().Gather()
	if err != nil {
		t.Fatal(err)
	}
	for _, family := range families {
		if family.GetName() != "spindle_workflow_duration_seconds" {
			continue
		}
		for _, m := range family.GetMetric() {
			if sum := m.GetHistogram().GetSampleSum(); sum > 0.65 {
				t.Errorf("recorded %vs for two 250ms runs, want about 0.5s", sum)
			}
		}
	}
}