package engine

import (
	"context"
	"errors"
	"log/slog"
	"os"
	"path/filepath"
	"reflect"
	"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/api/tangled"
	"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"
)

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
	runStepCalls    []models.WorkflowId
	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 tangled.Pipeline_Workflow, tpl tangled.Pipeline) (*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)
	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)
	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{
		Knot: "test-knot",
		Rkey: "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.Status != string(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.Status != string(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.Status != string(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{
		Knot: "test-knot",
		Rkey: "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.Status != string(models.StatusKindCancelled) {
		t.Fatalf("expected status to be cancelled, got %s", st.Status)
	}
}

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{Knot: "test-knot", Rkey: "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.Status != string(models.StatusKindFailed) {
		t.Fatalf("expected failed status, got %s", status.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 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{Knot: "test-knot", Rkey: "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.Status != string(models.StatusKindTimeout) {
		t.Fatalf("expected status to be timeout, got %s", st.Status)
	}

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

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{
		Knot: "test-knot",
		Rkey: "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.AtUri().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.AtUri().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{
		Knot: "test-knot",
		Rkey: "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"))
	}

	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{Knot: "test-knot", Rkey: "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)
	}
}