package engine

import (
	"context"
	"encoding/json"
	"errors"
	"log/slog"
	"os"
	"path/filepath"
	"reflect"
	"slices"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/bluesky-social/indigo/atproto/syntax"
	"go.opentelemetry.io/otel"
	sdktrace "go.opentelemetry.io/otel/sdk/trace"
	"go.opentelemetry.io/otel/sdk/trace/tracetest"
	"tangled.org/core/spindle/config"
	"tangled.org/core/spindle/db"
	"tangled.org/core/spindle/models"
	"tangled.org/core/spindle/observability"
	"tangled.org/core/spindle/quota"
	"tangled.org/core/spindle/secrets"
	"tangled.org/core/spindle/storage"
	"tangled.org/core/workflow"
)

type mockStep struct {
	name    string
	command string
}

func (m mockStep) Name() string          { return m.name }
func (m mockStep) Command() string       { return m.command }
func (m mockStep) Kind() models.StepKind { return models.StepKindUser }

type emptyVault struct{}

func (emptyVault) AddSecret(context.Context, secrets.UnlockedSecret) error {
	return nil
}

func (emptyVault) RemoveSecret(context.Context, secrets.Secret[any]) error {
	return nil
}

func (emptyVault) RemoveAllSecrets(context.Context, secrets.RepoIdentifier) error {
	return nil
}

func (emptyVault) GetSecretsLocked(context.Context, secrets.RepoIdentifier) ([]secrets.LockedSecret, error) {
	return nil, nil
}

func (emptyVault) GetSecretsUnlocked(context.Context, secrets.RepoIdentifier) ([]secrets.UnlockedSecret, error) {
	return nil, nil
}

type staticCacheController struct {
	bindings []models.CacheBinding
}

func (c staticCacheController) Plan(context.Context, *models.Pipeline, *models.Workflow) ([]models.CacheBinding, error) {
	return c.bindings, nil
}

func (staticCacheController) Apply(context.Context, CacheUpdate) error {
	return nil
}

type mockEngine struct {
	mu              sync.Mutex
	setupCalls      []models.WorkflowId
	setupTrusted    []bool
	runStepCalls    []models.WorkflowId
	runStepSecrets  [][]secrets.UnlockedSecret
	setupFunc       func(ctx context.Context, wid models.WorkflowId) error
	runStepFunc     func(ctx context.Context, wid models.WorkflowId, idx int, wfLogger models.WorkflowLogger) error
	destroyFunc     func(ctx context.Context, wid models.WorkflowId) error
	runWorkflowFunc func(*models.Workflow)
	timeout         time.Duration
}

func (m *mockEngine) InitWorkflow(twf workflow.WorkflowSpec, tpl workflow.PipelineSpec) (*models.Workflow, error) {
	owner := "did:web:alice"
	repo := "did:web:alice/repo"
	if md := tpl.TriggerMetadata; md != nil && md.Repo != nil {
		owner = md.Repo.Did
		if md.Repo.RepoDid != nil {
			repo = *md.Repo.RepoDid
		}
	}
	return &models.Workflow{
		OwnerDID: owner,
		RepoDID:  repo,
	}, nil
}

func (m *mockEngine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, wfLogger models.WorkflowLogger) error {
	m.mu.Lock()
	m.setupCalls = append(m.setupCalls, wid)
	m.setupTrusted = append(m.setupTrusted, wf.TrustedSource)
	fn := m.setupFunc
	m.mu.Unlock()
	if fn != nil {
		return fn(ctx, wid)
	}
	return nil
}

func (m *mockEngine) WorkflowTimeout() time.Duration {
	if m.timeout != 0 {
		return m.timeout
	}
	return 5 * time.Second
}

func (m *mockEngine) DestroyWorkflow(ctx context.Context, wid models.WorkflowId) error {
	if m.destroyFunc != nil {
		return m.destroyFunc(ctx, wid)
	}
	return nil
}

func (m *mockEngine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.Workflow, idx int, secrets []secrets.UnlockedSecret, wfLogger models.WorkflowLogger) error {
	m.mu.Lock()
	m.runStepCalls = append(m.runStepCalls, wid)
	m.runStepSecrets = append(m.runStepSecrets, slices.Clone(secrets))
	runStepFn := m.runStepFunc
	runWorkflowFn := m.runWorkflowFunc
	m.mu.Unlock()

	if runWorkflowFn != nil {
		runWorkflowFn(w)
	}
	if runStepFn != nil {
		return runStepFn(ctx, wid, idx, wfLogger)
	}
	return nil
}

func newTestDB(t *testing.T) *db.DB {
	t.Helper()
	d, err := db.Make(context.Background(), filepath.Join(t.TempDir(), "spindle.db"))
	if err != nil {
		t.Fatalf("failed to create test db: %v", err)
	}
	return d
}

func TestStartWorkflows_CollisionRejection(t *testing.T) {
	t.Parallel()

	testDB := newTestDB(t)
	logger := slog.New(slog.NewTextHandler(os.Stderr, nil))

	eng := &mockEngine{}
	pipelineId := models.PipelineId("test-rkey")

	// two names that normalize to the same wid must not both run
	wfColliding1 := models.Workflow{
		Name:  "test-job",
		Steps: []models.Step{mockStep{name: "step1"}},
	}
	wfColliding2 := models.Workflow{
		Name:  "test job",
		Steps: []models.Step{mockStep{name: "step1"}},
	}
	wfUnique := models.Workflow{
		Name:  "unique_job",
		Steps: []models.Step{mockStep{name: "step1"}},
	}

	pipeline := &models.Pipeline{
		Workflows: map[models.Engine][]models.Workflow{
			eng: {wfColliding1, wfColliding2, wfUnique},
		},
	}

	cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}}
	StartWorkflows(logger, nil, cfg, nil, nil, testDB, nil, nil, nil, context.Background(), pipeline, pipelineId)

	eng.mu.Lock()
	setupCalls := append([]models.WorkflowId(nil), eng.setupCalls...)
	eng.mu.Unlock()

	for _, call := range setupCalls {
		if call.Name == "test-job" || call.Name == "test job" {
			t.Fatalf("expected colliding workflow %s to not be started", call.Name)
		}
	}

	hasUnique := false
	for _, call := range setupCalls {
		if call.Name == "unique_job" {
			hasUnique = true
		}
	}
	if !hasUnique {
		t.Fatalf("expected unique workflow unique_job to be started")
	}

	widColliding1 := models.WorkflowId{PipelineId: pipelineId, Name: "test-job"}
	widColliding2 := models.WorkflowId{PipelineId: pipelineId, Name: "test job"}
	widUnique := models.WorkflowId{PipelineId: pipelineId, Name: "unique_job"}

	status1, err := testDB.GetStatus(widColliding1)
	if err != nil || status1 != models.StatusKindFailed {
		t.Fatalf("expected colliding1 status to be failed, got status=%v err=%v", status1, err)
	}

	status2, err := testDB.GetStatus(widColliding2)
	if err != nil || status2 != models.StatusKindFailed {
		t.Fatalf("expected colliding2 status to be failed, got status=%v err=%v", status2, err)
	}

	statusUnique, err := testDB.GetStatus(widUnique)
	if err != nil || statusUnique != models.StatusKindSuccess {
		t.Fatalf("expected unique status to be success, got status=%v err=%v", statusUnique, err)
	}
}

func TestCancelWorkflow_NotOverwritten(t *testing.T) {
	t.Parallel()

	testDB := newTestDB(t)
	logger := slog.New(slog.NewTextHandler(os.Stderr, nil))

	stepStarted := make(chan struct{})
	eng := &mockEngine{
		runStepFunc: func(ctx context.Context, wid models.WorkflowId, idx int, wfLogger models.WorkflowLogger) error {
			close(stepStarted)
			<-ctx.Done()
			return ctx.Err()
		},
	}

	pipelineId := models.PipelineId("test-rkey")

	wid := models.WorkflowId{
		PipelineId: pipelineId,
		Name:       "cancel_test_job",
	}

	pipeline := &models.Pipeline{
		Workflows: map[models.Engine][]models.Workflow{
			eng: {
				{
					Name:  "cancel_test_job",
					Steps: []models.Step{mockStep{name: "step1"}},
				},
			},
		},
	}

	cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}}
	doneChan := make(chan struct{})
	go func() {
		StartWorkflows(logger, nil, cfg, nil, nil, testDB, nil, nil, nil, context.Background(), pipeline, pipelineId)
		close(doneChan)
	}()

	select {
	case <-stepStarted:
	case <-time.After(5 * time.Second):
		t.Fatal("timed out waiting for step to start")
	}

	_ = testDB.StatusCancelled(wid, "User canceled the workflow", -1, nil)
	CancelWorkflow(wid)

	select {
	case <-doneChan:
	case <-time.After(5 * time.Second):
		t.Fatal("timed out waiting for StartWorkflows to complete")
	}

	// the runner writes StatusCancelled itself when it sees the canceled ctx
	// the handler writes nothing for a live wf, so nothing lands after to overwrite it
	st, err := testDB.GetStatus(wid)
	if err != nil {
		t.Fatalf("GetStatus error = %v", err)
	}
	if st != models.StatusKindCancelled {
		t.Fatalf("expected status to be cancelled, got %s", st)
	}
}

func TestStartWorkflowsScopesSecretsToTrustedPipelineRepo(t *testing.T) {
	vault, err := secrets.NewSQLiteManager(":memory:")
	if err != nil {
		t.Fatal(err)
	}
	repoA := syntax.DID("did:plc:workflowrepoa")
	repoB := syntax.DID("did:plc:workflowrepob")
	for _, secret := range []secrets.UnlockedSecret{
		{Repo: secrets.RepoIdentifier(repoA), Key: "A_SECRET", Value: "alpha"},
		{Repo: secrets.RepoIdentifier(repoB), Key: "B_SECRET", Value: "bravo"},
	} {
		if err := vault.AddSecret(context.Background(), secret); err != nil {
			t.Fatal(err)
		}
	}
	run := func(trusted bool) [][]secrets.UnlockedSecret {
		t.Helper()
		database := newTestDB(t)
		defer database.Close()
		engine := &mockEngine{}
		pipeline := &models.Pipeline{
			RepoDid:       repoA,
			TrustedSource: trusted,
			Workflows: map[models.Engine][]models.Workflow{
				engine: {{Name: "ci.yml", Steps: []models.Step{mockStep{name: "test"}}}},
			},
		}
		StartWorkflows(slog.Default(), vault, &config.Config{Server: config.Server{LogDir: t.TempDir()}}, nil, nil, database, nil, nil, nil, context.Background(), pipeline, models.PipelineId("3mu2xwiorc2xl"))
		engine.mu.Lock()
		defer engine.mu.Unlock()
		return slices.Clone(engine.runStepSecrets)
	}

	trustedSecrets := run(true)
	if len(trustedSecrets) != 1 || len(trustedSecrets[0]) != 1 || trustedSecrets[0][0].Key != "A_SECRET" || trustedSecrets[0][0].Value != "alpha" {
		t.Fatalf("trusted secrets = %#v", trustedSecrets)
	}
	if untrustedSecrets := run(false); len(untrustedSecrets) != 1 || len(untrustedSecrets[0]) != 0 {
		t.Fatalf("untrusted secrets = %#v", untrustedSecrets)
	}
}

func TestStartWorkflowsMarksOnlyTrustedWorkflowsTrusted(t *testing.T) {
	for _, tc := range []struct {
		name    string
		repoDid syntax.DID
		trusted bool
		want    bool
	}{
		{"trusted push", "did:plc:repo", true, true},
		{"untrusted source", "did:plc:repo", false, false},
		{"trusted source without a repo did", "", true, false},
	} {
		t.Run(tc.name, func(t *testing.T) {
			database := newTestDB(t)
			defer database.Close()
			vault, err := secrets.NewSQLiteManager(":memory:")
			if err != nil {
				t.Fatal(err)
			}
			eng := &mockEngine{}
			pipeline := &models.Pipeline{
				RepoDid:       tc.repoDid,
				TrustedSource: tc.trusted,
				Workflows: map[models.Engine][]models.Workflow{
					eng: {{Name: "ci.yml", Steps: []models.Step{mockStep{name: "test"}}}},
				},
			}
			StartWorkflows(slog.Default(), vault, &config.Config{Server: config.Server{LogDir: t.TempDir()}}, nil, nil, database, nil, nil, nil, context.Background(), pipeline, models.PipelineId("3mu2xwiorc2xl"))
			eng.mu.Lock()
			defer eng.mu.Unlock()
			if !slices.Equal(eng.setupTrusted, []bool{tc.want}) {
				t.Fatalf("setup saw TrustedSource %v, want [%t]", eng.setupTrusted, tc.want)
			}
		})
	}
}

func TestStartWorkflows_FlushesLogBeforeTerminalStatus(t *testing.T) {
	t.Parallel()

	testDB := newTestDB(t)
	logger := slog.New(slog.NewTextHandler(os.Stderr, nil))
	logDir := t.TempDir()
	vault, err := secrets.NewSQLiteManager(filepath.Join(t.TempDir(), "secrets.db"))
	if err != nil {
		t.Fatal(err)
	}

	repoDid := syntax.DID("did:plc:test")
	if err := vault.AddSecret(context.Background(), secrets.UnlockedSecret{
		Key:   "LONG_SECRET",
		Value: strings.Repeat("s", 256),
		Repo:  secrets.RepoIdentifier(repoDid.String()),
	}); err != nil {
		t.Fatal(err)
	}

	destroyStarted := make(chan struct{})
	releaseDestroy := make(chan struct{})
	var releaseOnce sync.Once
	release := func() {
		releaseOnce.Do(func() {
			close(releaseDestroy)
		})
	}
	t.Cleanup(release)

	eng := &mockEngine{
		runStepFunc: func(ctx context.Context, wid models.WorkflowId, idx int, wfLogger models.WorkflowLogger) error {
			if _, err := wfLogger.DataWriter(idx, "stdout").Write([]byte("ssh debug hint\n")); err != nil {
				return err
			}
			return errors.New("step failed")
		},
		destroyFunc: func(ctx context.Context, wid models.WorkflowId) error {
			close(destroyStarted)
			<-releaseDestroy
			return nil
		},
	}

	pipelineId := models.PipelineId("test-rkey")
	wid := models.WorkflowId{PipelineId: pipelineId, Name: "failed_job"}
	pipeline := &models.Pipeline{
		RepoDid:       repoDid,
		TrustedSource: true,
		Workflows: map[models.Engine][]models.Workflow{
			eng: {{Name: wid.Name, Steps: []models.Step{mockStep{name: "step1"}}}},
		},
	}
	cfg := &config.Config{Server: config.Server{LogDir: logDir}}

	done := make(chan struct{})
	go func() {
		StartWorkflows(logger, vault, cfg, nil, nil, testDB, nil, nil, nil, context.Background(), pipeline, pipelineId)
		close(done)
	}()

	select {
	case <-destroyStarted:
	case <-time.After(5 * time.Second):
		t.Fatal("timed out waiting for workflow destruction")
	}

	status, err := testDB.GetStatus(wid)
	if err != nil {
		t.Fatal(err)
	}
	if status != models.StatusKindFailed {
		t.Fatalf("expected failed status, got %s", status)
	}

	log, err := os.ReadFile(models.LogFilePath(logDir, wid))
	if err != nil {
		t.Fatal(err)
	}
	if !strings.Contains(string(log), "ssh debug hint") {
		t.Fatalf("terminal status was visible before the log was flushed: %s", log)
	}

	release()
	select {
	case <-done:
	case <-time.After(5 * time.Second):
		t.Fatal("timed out waiting for StartWorkflows to complete")
	}
}

func TestStartWorkflows_EndLinesCarryStepOutcome(t *testing.T) {
	t.Parallel()

	testDB := newTestDB(t)
	logger := slog.New(slog.NewTextHandler(os.Stderr, nil))
	logDir := t.TempDir()

	eng := &mockEngine{
		runStepFunc: func(ctx context.Context, wid models.WorkflowId, idx int, wfLogger models.WorkflowLogger) error {
			if idx == 1 {
				hint := mockStep{name: "Debug shell"}
				wfLogger.ControlWriter(2, hint, models.StepStatusStart).Write([]byte{0})
				wfLogger.ControlWriter(2, hint, models.StepStatusEnd).Write([]byte{0})
				return errors.New("exited with code 1")
			}
			return nil
		},
	}

	pipelineId := models.PipelineId("outcome-rkey")
	wid := models.WorkflowId{PipelineId: pipelineId, Name: "outcome"}
	pipeline := &models.Pipeline{
		Workflows: map[models.Engine][]models.Workflow{
			eng: {{Name: wid.Name, Steps: []models.Step{mockStep{name: "build"}, mockStep{name: "test"}}}},
		},
	}
	cfg := &config.Config{Server: config.Server{LogDir: logDir}}
	StartWorkflows(logger, emptyVault{}, cfg, nil, nil, testDB, nil, nil, nil, context.Background(), pipeline, pipelineId)

	raw, err := os.ReadFile(models.LogFilePath(logDir, wid))
	if err != nil {
		t.Fatal(err)
	}
	got := map[string]models.StepOutcome{}
	for _, encoded := range strings.Split(strings.TrimSpace(string(raw)), "\n") {
		var line models.LogLine
		if err := json.Unmarshal([]byte(encoded), &line); err != nil {
			t.Fatalf("decode log line %q: %v", encoded, err)
		}
		if line.Kind == models.LogKindControl && line.StepStatus == models.StepStatusEnd {
			got[line.Content] = line.StepOutcome
		}
	}
	want := map[string]models.StepOutcome{
		"build":       models.StepOutcomeSuccess,
		"test":        models.StepOutcomeFailed,
		"Debug shell": "",
	}
	if !reflect.DeepEqual(got, want) {
		t.Fatalf("end line outcomes = %v, want %v", got, want)
	}
}

func TestStartWorkflowsCapsTheWorkflowTimeBudget(t *testing.T) {
	deadline := make(chan time.Time, 1)
	eng := &mockEngine{
		timeout: 24 * time.Hour,
		setupFunc: func(ctx context.Context, _ models.WorkflowId) error {
			got, ok := ctx.Deadline()
			if !ok {
				return errors.New("workflow context has no deadline")
			}
			deadline <- got
			return nil
		},
	}
	pipeline := &models.Pipeline{Workflows: map[models.Engine][]models.Workflow{
		eng: {{Name: "bounded", Steps: []models.Step{mockStep{name: "step"}}}},
	}}

	StartWorkflows(slog.Default(), nil, &config.Config{Server: config.Server{LogDir: t.TempDir()}}, nil, nil, newTestDB(t), nil, nil, nil, context.Background(), pipeline, "bounded")

	remaining := time.Until(<-deadline)
	if remaining > models.MaxWorkflowTimeBudget || remaining < models.MaxWorkflowTimeBudget-time.Minute {
		t.Fatalf("workflow time budget = %s, want %s", remaining, models.MaxWorkflowTimeBudget)
	}
}

func TestSetupTimeout_ReportsTimeout(t *testing.T) {
	t.Parallel()

	testDB := newTestDB(t)
	logger := slog.New(slog.NewTextHandler(os.Stderr, nil))

	// setup blocks past the workflow timeout, so it should land as timeout not failed
	eng := &mockEngine{
		timeout: 100 * time.Millisecond,
		setupFunc: func(ctx context.Context, wid models.WorkflowId) error {
			<-ctx.Done()
			return ctx.Err()
		},
	}

	pipelineId := models.PipelineId("test-rkey")
	wid := models.WorkflowId{PipelineId: pipelineId, Name: "timeout_job"}

	pipeline := &models.Pipeline{
		Workflows: map[models.Engine][]models.Workflow{
			eng: {{Name: "timeout_job", Steps: []models.Step{mockStep{name: "step1"}}}},
		},
	}

	cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}}
	StartWorkflows(logger, nil, cfg, nil, nil, testDB, nil, nil, nil, context.Background(), pipeline, pipelineId)

	st, err := testDB.GetStatus(wid)
	if err != nil {
		t.Fatalf("GetStatus error = %v", err)
	}
	if st != models.StatusKindTimeout {
		t.Fatalf("expected status to be timeout, got %s", st)
	}

	if len(eng.runStepCalls) != 0 {
		t.Fatalf("expected no steps to run after setup timeout, got %d", len(eng.runStepCalls))
	}
}

type cacheSave struct {
	keys     []string
	ctxErr   error
	deadline time.Time
}

type cacheMockEngine struct {
	*mockEngine
	blockSave bool
	mu        sync.Mutex
	saves     []cacheSave
	events    []string
}

func (e *cacheMockEngine) record(event string) {
	e.mu.Lock()
	defer e.mu.Unlock()
	e.events = append(e.events, event)
}

func (e *cacheMockEngine) RestoreCache(context.Context, models.WorkflowId, *models.Workflow, storage.Storage, []models.CacheBinding, models.WorkflowLogger) error {
	return nil
}

func (e *cacheMockEngine) SaveCache(ctx context.Context, _ models.WorkflowId, _ *models.Workflow, _ storage.Storage, caches []models.CacheBinding, _ models.WorkflowLogger) error {
	save := cacheSave{ctxErr: ctx.Err()}
	save.deadline, _ = ctx.Deadline()
	for _, c := range caches {
		save.keys = append(save.keys, c.Key)
	}
	if e.blockSave {
		<-ctx.Done()
	}
	e.mu.Lock()
	e.saves = append(e.saves, save)
	e.mu.Unlock()
	e.record("save")
	return ctx.Err()
}

type cachedRun struct {
	status models.StatusKind
	saves  []cacheSave
	events []string
	log    string
}

func runCachedWorkflow(t *testing.T, pipelineID models.PipelineId, blockSave bool, runStep func(ctx context.Context, wid models.WorkflowId) error) cachedRun {
	t.Helper()
	eng := &cacheMockEngine{blockSave: blockSave}
	eng.mockEngine = &mockEngine{
		timeout: 100 * time.Millisecond,
		runStepFunc: func(ctx context.Context, wid models.WorkflowId, _ int, _ models.WorkflowLogger) error {
			return runStep(ctx, wid)
		},
		destroyFunc: func(context.Context, models.WorkflowId) error {
			eng.record("destroy")
			return nil
		},
	}
	wid := models.WorkflowId{PipelineId: pipelineID, Name: "cached"}
	pipeline := &models.Pipeline{
		RepoDid:       "did:plc:cached",
		TrustedSource: true,
		Workflows: map[models.Engine][]models.Workflow{
			eng: {{
				Name:  wid.Name,
				Steps: []models.Step{mockStep{name: "build"}},
				Caches: []models.CacheEntry{
					{Key: "always", Paths: []string{"a"}, When: "always"},
					{Key: "on-success", Paths: []string{"b"}},
				},
			}},
		},
	}
	controller := staticCacheController{bindings: []models.CacheBinding{
		{EntryIndex: 0, Key: "always", SaveKey: "objects/always", When: "always"},
		{EntryIndex: 1, Key: "on-success", SaveKey: "objects/on-success"},
	}}
	testDB := newTestDB(t)
	cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}}
	StartWorkflows(discardLogger, emptyVault{}, cfg, nil, nil, testDB, nil, &fakeStorage{objects: map[string][]byte{}}, controller, context.Background(), pipeline, pipelineID)

	status, err := testDB.GetStatus(wid)
	if err != nil {
		t.Fatal(err)
	}
	log, err := os.ReadFile(models.LogFilePath(cfg.Server.LogDir, wid))
	if err != nil {
		t.Fatal(err)
	}
	eng.mu.Lock()
	defer eng.mu.Unlock()
	return cachedRun{status: status, saves: eng.saves, events: eng.events, log: string(log)}
}

var timeoutSteps = map[string]func(context.Context, models.WorkflowId) error{
	"deadline passed": func(ctx context.Context, _ models.WorkflowId) error {
		<-ctx.Done()
		return ErrTimedOut
	},
	"engine reported": func(context.Context, models.WorkflowId) error {
		return ErrTimedOut
	},
}

func timeoutPipelineID(prefix, name string) models.PipelineId {
	return models.PipelineId(prefix + "-" + strings.ReplaceAll(name, " ", "-"))
}

func TestStartWorkflowsSavesAlwaysCachesAfterTimeout(t *testing.T) {
	t.Parallel()
	for name, runStep := range timeoutSteps {
		t.Run(name, func(t *testing.T) {
			t.Parallel()
			start := time.Now()
			run := runCachedWorkflow(t, timeoutPipelineID("timeout", name), false, runStep)
			if run.status != models.StatusKindTimeout {
				t.Fatalf("status = %s, want timeout", run.status)
			}
			if len(run.saves) != 1 || !slices.Equal(run.saves[0].keys, []string{"always"}) {
				t.Fatalf("saves = %+v, want only the always cache", run.saves)
			}
			if run.saves[0].ctxErr != nil {
				t.Fatalf("save context already done: %v", run.saves[0].ctxErr)
			}
			if run.saves[0].deadline.IsZero() || run.saves[0].deadline.After(time.Now().Add(CacheSaveTimeout)) || run.saves[0].deadline.Before(start.Add(CacheSaveTimeout)) {
				t.Fatalf("save deadline = %v, want %s after the timeout", run.saves[0].deadline, CacheSaveTimeout)
			}
			if !slices.Equal(run.events, []string{"save", "destroy"}) {
				t.Fatalf("events = %v, want save then a single destroy", run.events)
			}
			if !strings.Contains(run.log, CacheSaveStep.Name()) {
				t.Fatalf("log has no %q step: %s", CacheSaveStep.Name(), run.log)
			}
		})
	}
}

func TestStartWorkflowsDestroysAfterABlockedSave(t *testing.T) {
	saveTimeout := CacheSaveTimeout
	CacheSaveTimeout = 200 * time.Millisecond
	t.Cleanup(func() { CacheSaveTimeout = saveTimeout })

	for name, runStep := range timeoutSteps {
		t.Run(name, func(t *testing.T) {
			start := time.Now()
			run := runCachedWorkflow(t, timeoutPipelineID("blocked", name), true, runStep)
			if run.status != models.StatusKindTimeout {
				t.Fatalf("status = %s, want timeout", run.status)
			}
			if !slices.Equal(run.events, []string{"save", "destroy"}) {
				t.Fatalf("events = %v, want save then a single destroy", run.events)
			}
			if took := time.Since(start); took > 5*time.Second {
				t.Fatalf("workflow took %v with a blocked save", took)
			}
		})
	}
}

func TestStartWorkflowsSavesNothingAfterCancel(t *testing.T) {
	t.Parallel()
	run := runCachedWorkflow(t, "cancel-cached", false, func(ctx context.Context, wid models.WorkflowId) error {
		CancelWorkflow(wid)
		<-ctx.Done()
		return ErrWorkflowCanceled
	})
	if run.status != models.StatusKindCancelled {
		t.Fatalf("status = %s, want cancelled", run.status)
	}
	if !slices.Equal(run.events, []string{"destroy"}) {
		t.Fatalf("events = %v, want only a destroy", run.events)
	}
}

func TestStartWorkflowsSavesAlwaysCachesAfterFailure(t *testing.T) {
	t.Parallel()
	run := runCachedWorkflow(t, "failed-cached", false, func(context.Context, models.WorkflowId) error {
		return errors.New("exited with code 1")
	})
	if run.status != models.StatusKindFailed {
		t.Fatalf("status = %s, want failed", run.status)
	}
	if len(run.saves) != 1 || !slices.Equal(run.saves[0].keys, []string{"always"}) {
		t.Fatalf("saves = %+v, want only the always cache", run.saves)
	}
	if !slices.Equal(run.events, []string{"save", "destroy"}) {
		t.Fatalf("events = %v, want save then a single destroy", run.events)
	}
}

func TestStartWorkflows_SpanEnrichment(t *testing.T) {
	oldProvider := otel.GetTracerProvider()
	recorder := tracetest.NewSpanRecorder()
	provider := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(recorder))
	otel.SetTracerProvider(provider)
	t.Cleanup(func() {
		_ = provider.Shutdown(context.Background())
		otel.SetTracerProvider(oldProvider)
	})

	testDB := newTestDB(t)
	logger := slog.New(slog.NewTextHandler(os.Stderr, nil))

	eng := &mockEngine{}
	pipelineId := models.PipelineId("test-rkey")

	pipeline := &models.Pipeline{
		RepoDid: "did:web:test-repo",
		Workflows: map[models.Engine][]models.Workflow{
			eng: {
				{
					Name: "test-workflow",
					Steps: []models.Step{
						mockStep{name: "step-1"},
					},
					OwnerDID: "did:web:test-owner",
					RepoDID:  "did:web:test-repo",
				},
			},
		},
		TrustedSource: false,
	}

	cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}}
	StartWorkflows(logger, nil, cfg, nil, nil, testDB, nil, nil, nil, context.Background(), pipeline, pipelineId)

	spans := recorder.Ended()
	var workflowRunSpan, stepRunSpan sdktrace.ReadOnlySpan
	for _, span := range spans {
		if span.Name() == "workflow.run" {
			workflowRunSpan = span
		} else if span.Name() == "step.run" {
			stepRunSpan = span
		}
	}

	if workflowRunSpan == nil {
		t.Fatal("workflow.run span not recorded")
	}
	if stepRunSpan == nil {
		t.Fatal("step.run span not recorded")
	}

	workflowAttrs := make(map[string]string)
	for _, attr := range workflowRunSpan.Attributes() {
		workflowAttrs[string(attr.Key)] = attr.Value.AsString()
	}

	wid := models.WorkflowId{
		PipelineId: pipelineId,
		Name:       "test-workflow",
	}

	expectedWfAttrs := map[string]string{
		observability.OwnerDIDKey:   "did:web:test-owner",
		observability.RepoDIDKey:    "did:web:test-repo",
		observability.PipelineIDKey: pipelineId.String(),
		observability.WorkflowIDKey: wid.String(),
	}
	for k, wantVal := range expectedWfAttrs {
		if gotVal := workflowAttrs[k]; gotVal != wantVal {
			t.Errorf("workflow attribute %q = %q, want %q", k, gotVal, wantVal)
		}
	}

	stepAttrs := make(map[string]string)
	for _, attr := range stepRunSpan.Attributes() {
		stepAttrs[string(attr.Key)] = attr.Value.AsString()
	}

	expectedStepAttrs := map[string]string{
		observability.OwnerDIDKey:   "did:web:test-owner",
		observability.RepoDIDKey:    "did:web:test-repo",
		observability.PipelineIDKey: pipelineId.String(),
		observability.WorkflowIDKey: wid.String(),
		observability.StepNameKey:   "step-1",
	}
	for k, wantVal := range expectedStepAttrs {
		if gotVal := stepAttrs[k]; gotVal != wantVal {
			t.Errorf("step attribute %q = %q, want %q", k, gotVal, wantVal)
		}
	}
	hasStepIndex := false
	for _, attr := range stepRunSpan.Attributes() {
		if string(attr.Key) == observability.StepIndexKey && attr.Value.AsInterface() == int64(0) {
			hasStepIndex = true
		}
	}
	if !hasStepIndex {
		t.Error("step.run span missing step_index attribute or it is not 0")
	}
}

type mockResourceUsageReporter struct {
	*mockEngine
	workflows, memory, vcpus, disk int64
	usage                          WorkflowResourceUsage
	usageOk                        bool
}

func (m *mockResourceUsageReporter) QuotaResources(wf *models.Workflow) quota.Resources {
	return quota.Resources{
		quota.ResourceWorkflows: m.workflows,
		quota.ResourceMemoryMiB: m.memory,
		quota.ResourceVCPUs:     m.vcpus,
		quota.ResourceDiskMiB:   m.disk,
	}
}

func (m *mockResourceUsageReporter) WorkflowResourceUsage(wf *models.Workflow) (WorkflowResourceUsage, bool) {
	return m.usage, m.usageOk
}

func TestStartWorkflows_ResourceTracking(t *testing.T) {
	oldProvider := otel.GetTracerProvider()
	recorder := tracetest.NewSpanRecorder()
	provider := sdktrace.NewTracerProvider(
		sdktrace.WithSpanProcessor(recorder),
		sdktrace.WithSampler(sdktrace.AlwaysSample()),
	)
	otel.SetTracerProvider(provider)
	t.Cleanup(func() {
		_ = provider.Shutdown(context.Background())
		otel.SetTracerProvider(oldProvider)
	})

	testDB := newTestDB(t)

	mu := &sync.Mutex{}
	records := []slog.Record{}
	handler := &captureHandler{
		mu:      mu,
		records: &records,
	}
	logger := slog.New(handler)

	eng := &mockResourceUsageReporter{
		mockEngine: &mockEngine{},
		workflows:  2,
		memory:     1024,
		vcpus:      4,
		disk:       512,
		usage: WorkflowResourceUsage{
			CPUUsec:              12345,
			MemoryCurrentBytes:   2048,
			MemoryPeakBytes:      4096,
			SwapCurrentBytes:     100,
			SwapPeakBytes:        200,
			PIDsCurrent:          10,
			IOReadBytes:          300,
			IOWriteBytes:         400,
			IOReadOps:            30,
			IOWriteOps:           40,
			VolumeAllocatedBytes: 1000,
			CgroupAvailable:      true,
			VolumeAvailable:      true,
		},
		usageOk: true,
	}

	pipelineId := models.PipelineId("test-rkey")

	pipeline := &models.Pipeline{
		RepoDid: "did:web:test-repo",
		Workflows: map[models.Engine][]models.Workflow{
			eng: {
				{
					Name: "test-workflow",
					Steps: []models.Step{
						mockStep{name: "step-1"},
					},
					OwnerDID: "did:web:test-owner",
					RepoDID:  "did:web:test-repo",
				},
			},
		},
		TrustedSource: false,
	}

	cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}}

	metrics := observability.NewMetrics()
	ctx := observability.WithMetrics(context.Background(), metrics)

	StartWorkflows(logger, nil, cfg, nil, nil, testDB, nil, nil, nil, ctx, pipeline, pipelineId)

	handler.mu.Lock()
	records = *handler.records
	handler.mu.Unlock()

	var startedLog, finishedLog slog.Record
	for _, r := range records {
		if r.Message == "workflow started" {
			startedLog = r
		} else if r.Message == "workflow finished" {
			finishedLog = r
		}
	}

	if startedLog.Message == "" {
		t.Fatal("workflow started log not found")
	}
	if finishedLog.Message == "" {
		t.Fatal("workflow finished log not found")
	}

	getAttr := func(r slog.Record, key string) any {
		var val any
		r.Attrs(func(a slog.Attr) bool {
			if a.Key == key {
				val = a.Value.Any()
				return false
			}
			return true
		})
		return val
	}

	if getAttr(startedLog, "requested_workflows") != int64(2) {
		t.Errorf("started log requested_workflows = %v, want 2", getAttr(startedLog, "requested_workflows"))
	}
	if getAttr(startedLog, "requested_vcpus") != int64(4) {
		t.Errorf("started log requested_vcpus = %v, want 4", getAttr(startedLog, "requested_vcpus"))
	}

	if getAttr(finishedLog, "result") != "success" {
		t.Errorf("finished log result = %v, want success", getAttr(finishedLog, "result"))
	}
	if getAttr(finishedLog, "actual_cpu_usec") != uint64(12345) {
		t.Errorf("finished log actual_cpu_usec = %v, want 12345", getAttr(finishedLog, "actual_cpu_usec"))
	}
	if getAttr(finishedLog, "actual_memory_peak_bytes") != uint64(4096) {
		t.Errorf("finished log actual_memory_peak_bytes = %v, want 4096", getAttr(finishedLog, "actual_memory_peak_bytes"))
	}
	if getAttr(finishedLog, "error") != "" || getAttr(finishedLog, "failure_class") != string(FailureClassNone) {
		t.Errorf("finished log error = %q, failure_class = %q, want no error", getAttr(finishedLog, "error"), getAttr(finishedLog, "failure_class"))
	}
	duration, _ := getAttr(finishedLog, "duration_seconds").(float64)
	waited, _ := getAttr(finishedLog, "wait_seconds").(float64)
	if waited < 0 || duration < 0 {
		t.Errorf("finished log wait %v, duration %v, want both >= 0", waited, duration)
	}

	spans := recorder.Ended()
	var workflowRunSpan sdktrace.ReadOnlySpan
	for _, span := range spans {
		if span.Name() == "workflow.run" {
			workflowRunSpan = span
			break
		}
	}

	if workflowRunSpan == nil {
		t.Fatal("workflow.run span not found")
	}

	attrs := make(map[string]any)
	for _, attr := range workflowRunSpan.Attributes() {
		attrs[string(attr.Key)] = attr.Value.AsInterface()
	}

	if attrs[observability.ReqWorkflowsKey] != int64(2) {
		t.Errorf("span attr %s = %v, want 2", observability.ReqWorkflowsKey, attrs[observability.ReqWorkflowsKey])
	}
	if attrs[observability.ReqVCPUsKey] != int64(4) {
		t.Errorf("span attr %s = %v, want 4", observability.ReqVCPUsKey, attrs[observability.ReqVCPUsKey])
	}

	if attrs[observability.ActCPUUsecKey] != int64(12345) {
		t.Errorf("span attr %s = %v, want 12345", observability.ActCPUUsecKey, attrs[observability.ActCPUUsecKey])
	}
	if attrs[observability.ActMemoryPeakBytesKey] != int64(4096) {
		t.Errorf("span attr %s = %v, want 4096", observability.ActMemoryPeakBytesKey, attrs[observability.ActMemoryPeakBytesKey])
	}
}

type captureHandler struct {
	attrs   []slog.Attr
	records *[]slog.Record
	mu      *sync.Mutex
}

func (h *captureHandler) Enabled(ctx context.Context, level slog.Level) bool {
	return true
}

func (h *captureHandler) Handle(ctx context.Context, r slog.Record) error {
	h.mu.Lock()
	defer h.mu.Unlock()
	r.AddAttrs(h.attrs...)
	*h.records = append(*h.records, r)
	return nil
}

func (h *captureHandler) WithAttrs(attrs []slog.Attr) slog.Handler {
	newAttrs := append([]slog.Attr(nil), h.attrs...)
	newAttrs = append(newAttrs, attrs...)
	return &captureHandler{
		attrs:   newAttrs,
		records: h.records,
		mu:      h.mu,
	}
}

func (h *captureHandler) WithGroup(name string) slog.Handler {
	return h
}

func TestStartWorkflowsPassesPlannedCacheBindingsToEngine(t *testing.T) {
	t.Parallel()

	database := newTestDB(t)
	logger := slog.New(slog.NewTextHandler(os.Stderr, nil))
	expected := []models.CacheBinding{{
		EntryIndex: 0,
		Paths:      []string{"deps"},
		Key:        "deps",
		SaveID:     "cache-id",
		SaveKey:    "objects/cache-id",
	}}
	var got []models.CacheBinding
	eng := &mockEngine{
		runWorkflowFunc: func(workflow *models.Workflow) {
			got = append([]models.CacheBinding(nil), workflow.CacheBindings...)
		},
	}
	pipelineID := models.PipelineId("test-rkey")
	pipeline := &models.Pipeline{
		RepoDid:       "did:web:example.com",
		TrustedSource: true,
		Workflows: map[models.Engine][]models.Workflow{
			eng: {{
				Name:   "cache-job",
				Engine: "microvm",
				Steps:  []models.Step{mockStep{name: "step1"}},
				Caches: []models.CacheEntry{{Key: "deps", Paths: []string{"deps"}}},
			}},
		},
	}
	cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}}
	store := &fakeStorage{objects: make(map[string][]byte)}

	StartWorkflows(
		logger,
		emptyVault{},
		cfg,
		nil,
		nil,
		database,
		nil,
		store,
		staticCacheController{bindings: expected},
		context.Background(),
		pipeline,
		pipelineID,
	)

	if !reflect.DeepEqual(got, expected) {
		t.Fatalf("cache bindings passed to engine = %#v, want %#v", got, expected)
	}
}

func TestStartWorkflowsDestroysWorkflowWhenSetupIsCancelled(t *testing.T) {
	t.Parallel()

	testDB := newTestDB(t)
	logger := slog.New(slog.NewTextHandler(os.Stderr, nil))

	setupStarted := make(chan struct{})
	destroyed := make(chan struct{}, 1)
	eng := &mockEngine{
		setupFunc: func(ctx context.Context, wid models.WorkflowId) error {
			close(setupStarted)
			<-ctx.Done()
			return ctx.Err()
		},
		destroyFunc: func(_ context.Context, wid models.WorkflowId) error {
			destroyed <- struct{}{}
			return nil
		},
	}

	pipelineId := models.PipelineId("test-rkey")

	wid := models.WorkflowId{
		PipelineId: pipelineId,
		Name:       "cancel_setup_job",
	}

	pipeline := &models.Pipeline{
		Workflows: map[models.Engine][]models.Workflow{
			eng: {
				{
					Name:  "cancel_setup_job",
					Steps: []models.Step{mockStep{name: "step1"}},
				},
			},
		},
	}

	cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}}
	doneChan := make(chan struct{})
	go func() {
		StartWorkflows(logger, nil, cfg, nil, nil, testDB, nil, nil, nil, context.Background(), pipeline, pipelineId)
		close(doneChan)
	}()

	select {
	case <-setupStarted:
	case <-time.After(5 * time.Second):
		t.Fatal("timed out waiting for workflow setup to start")
	}

	CancelWorkflow(wid)

	select {
	case <-doneChan:
	case <-time.After(5 * time.Second):
		t.Fatal("timed out waiting for StartWorkflows to complete")
	}

	select {
	case <-destroyed:
	case <-time.After(5 * time.Second):
		t.Fatal("cancelled setup did not destroy the workflow")
	}

	st, err := testDB.GetStatus(wid)
	if err != nil {
		t.Fatalf("GetStatus error = %v", err)
	}
	if st != models.StatusKindCancelled {
		t.Fatalf("expected status to be cancelled, got %s", st)
	}
}