package engine

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"os"
	"slices"
	"sync"
	"time"

	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/codes"
	"go.opentelemetry.io/otel/trace"
	"strings"
	"tangled.org/core/notifier"
	"tangled.org/core/spindle/artifactstore"
	"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"
)

var (
	ErrTimedOut         = errors.New("timed out")
	ErrWorkflowFailed   = errors.New("workflow failed")
	ErrWorkflowCanceled = errors.New("workflow canceled")
)

type FailureClass string

const (
	FailureClassNone           FailureClass = "none"
	FailureClassUser           FailureClass = "user"
	FailureClassInfrastructure FailureClass = "infrastructure"
	FailureClassPolicy         FailureClass = "policy"
)

type FailureReason string

const (
	FailureReasonSuccess             FailureReason = "success"
	FailureReasonCommandFailed       FailureReason = "command_failed"
	FailureReasonResourceExhausted   FailureReason = "resource_exhausted"
	FailureReasonConfigurationFailed FailureReason = "configuration_failed"
	FailureReasonWorkflowInvalid     FailureReason = "workflow_invalid"
	FailureReasonSetupFailed         FailureReason = "setup_failed"
	FailureReasonRuntimeFailed       FailureReason = "runtime_failed"
	FailureReasonCapacityUnavailable FailureReason = "capacity_unavailable"
	FailureReasonQuotaDenied         FailureReason = "quota_denied"
	FailureReasonNoExecutor          FailureReason = "no_executor"
	FailureReasonExecutorLost        FailureReason = "executor_lost"
	FailureReasonTimeout             FailureReason = "timeout"
	FailureReasonCancelled           FailureReason = "cancelled"
)

type WorkflowFailure struct {
	Class  FailureClass
	Reason FailureReason
	Err    error
}

func (e *WorkflowFailure) Error() string {
	if e.Err == nil {
		return string(e.Reason)
	}
	return e.Err.Error()
}

func (e *WorkflowFailure) Unwrap() error {
	return e.Err
}

func ClassifiedFailure(class FailureClass, reason FailureReason, err error) error {
	if err == nil {
		return nil
	}
	var classified *WorkflowFailure
	if errors.As(err, &classified) {
		return err
	}
	return &WorkflowFailure{Class: class, Reason: reason, Err: err}
}

func FailureAttribution(result string, err error) (string, string) {
	switch result {
	case "success":
		return string(FailureClassNone), string(FailureReasonSuccess)
	case "timeout":
		return string(FailureClassUser), string(FailureReasonTimeout)
	case "cancelled":
		return string(FailureClassUser), string(FailureReasonCancelled)
	}

	var failure *WorkflowFailure
	if errors.As(err, &failure) {
		return string(failure.Class), string(failure.Reason)
	}
	if errors.Is(err, ErrNoWorkflowSlots) {
		return string(FailureClassInfrastructure), string(FailureReasonCapacityUnavailable)
	}
	return string(FailureClassInfrastructure), string(FailureReasonRuntimeFailed)
}

var (
	activeMu      sync.Mutex
	activeCancels = make(map[models.WorkflowId]context.CancelCauseFunc)
)

func CancelWorkflow(wid models.WorkflowId) {
	activeMu.Lock()
	cancel, ok := activeCancels[wid]
	activeMu.Unlock()
	if ok {
		cancel(ErrWorkflowCanceled)
	}
}

// user cancel, timeout is DeadlineExceeded
func isCanceled(wfCtx context.Context) bool {
	return errors.Is(context.Cause(wfCtx), ErrWorkflowCanceled)
}

// for when recording early wf cancellations
func writeWfError(db *db.DB, n *notifier.Notifier, l *slog.Logger, wfCtx context.Context, wid models.WorkflowId, phase string, err error) {
	l = l.With("wid", wid, "phase", phase)
	switch {
	case isCanceled(wfCtx):
		l.InfoContext(wfCtx, "workflow canceled")
		if dbErr := db.StatusCancelled(wid, "User canceled the workflow", -1, n); dbErr != nil {
			l.ErrorContext(wfCtx, "failed to set workflow status to cancelled", "err", dbErr)
		}
	case errors.Is(err, ErrTimedOut) || errors.Is(wfCtx.Err(), context.DeadlineExceeded):
		l.InfoContext(wfCtx, "workflow timed out")
		if dbErr := db.StatusTimeout(wid, n); dbErr != nil {
			l.ErrorContext(wfCtx, "failed to set workflow status to timeout", "err", dbErr)
		}
	default:
		l.ErrorContext(wfCtx, "workflow failed", "err", err)
		if dbErr := db.StatusFailed(wid, err.Error(), -1, n); dbErr != nil {
			l.ErrorContext(wfCtx, "failed to set workflow status to failed", "err", dbErr)
		}
	}
}

// the mill streams the executor's real log file into place itself
// a local logger here would only write competing lines
type workflowLoggerProvider interface {
	WorkflowLogger(wid models.WorkflowId) models.WorkflowLogger
}

// for engines that manage status updates outside StartWorkflows
type RemoteStatusEngine interface {
	AuthorsRemoteStatus()
}

type WorkflowQuotaReporter interface {
	QuotaResources(wf *models.Workflow) quota.Resources
}

type WorkflowQuotaStoreBinder interface {
	BindWorkflowQuotaStore(wf *models.Workflow, store quota.ReservationStore) error
}

// the executor has no repo rows to judge a debug key by, so it hands the engine a check that asks the mill
type WorkflowDebugKeyBinder interface {
	BindWorkflowDebugKeys(wf *models.Workflow, check func(ctx context.Context, key []byte) bool) error
}

type metricEngineNamer interface {
	MetricEngineName() string
}

func engineName(eng models.Engine) string {
	if eng == nil {
		return "unknown"
	}
	if named, ok := eng.(metricEngineNamer); ok {
		return boundEngineName(named.MetricEngineName())
	}
	name := strings.TrimPrefix(fmt.Sprintf("%T", eng), "*")
	if idx := strings.Index(name, "."); idx != -1 {
		name = name[:idx]
	}
	return boundEngineName(name)
}

func boundEngineName(name string) string {
	switch name {
	case "dummy", "microvm", "nixery":
		return name
	default:
		return "unknown"
	}
}

func workflowResult(ctx context.Context, err error) string {
	if isCanceled(ctx) {
		return "cancelled"
	}
	if errors.Is(err, ErrTimedOut) || errors.Is(ctx.Err(), context.DeadlineExceeded) {
		return "timeout"
	}
	if err != nil {
		return "failure"
	}
	return "success"
}

func truncateRunes(s string, n int) string {
	r := []rune(s)
	if len(r) <= n {
		return s
	}
	return string(r[:n]) + "…"
}

func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, qm *quota.Manager, stores *artifactstore.Stores, db *db.DB, n *notifier.Notifier, cacheStore storage.Storage, cacheController CacheController, ctx context.Context, pipeline *models.Pipeline, pipelineId models.PipelineId) {
	l.Info("starting all workflows in parallel", "pipeline", pipelineId)

	isTrustedRepo := pipeline.TrustedSource && pipeline.RepoDid != ""
	var allSecrets []secrets.UnlockedSecret
	// never pass secrets to pipelines that run untrusted (e.g. fork) code
	if isTrustedRepo {
		if res, err := vault.GetSecretsUnlocked(ctx, secrets.RepoIdentifier(pipeline.RepoDid.String())); err == nil {
			allSecrets = res
		}
	} else if !pipeline.TrustedSource {
		l.Info("skipping secrets for untrusted pipeline source", "pipeline", pipelineId)
	}
	cacheEnabled := cacheStore != nil && cacheController != nil && isTrustedRepo
	if cacheStore != nil && !pipeline.TrustedSource {
		l.Info("skipping caches for untrusted pipeline source", "pipeline", pipelineId)
	}

	secretValues := make([]string, len(allSecrets))
	for i, s := range allSecrets {
		secretValues[i] = s.Value
	}

	// wid.String() is lossy so two different names can map to the same key
	// eg. "foo bar" and "foo-bar"...
	wfCounts := make(map[string]int)
	for _, wfs := range pipeline.Workflows {
		for _, w := range wfs {
			wid := models.WorkflowId{
				PipelineId: pipelineId,
				Name:       w.Name,
			}
			wfCounts[wid.String()]++
		}
	}
	var wg sync.WaitGroup
	for eng, wfs := range pipeline.Workflows {
		workflowTimeout := models.WorkflowTimeBudget(eng.WorkflowTimeout())
		cacheRunner, cachesSupported := eng.(CacheRunner)
		l.Info("using workflow timeout", "timeout", workflowTimeout)

		for _, w := range wfs {
			w := w
			w.TrustedSource = isTrustedRepo
			wid := models.WorkflowId{
				PipelineId: pipelineId,
				Name:       w.Name,
			}

			if wfCounts[wid.String()] > 1 {
				l.Warn("skipping workflow due to name collision", "wid", wid, "key", wid.String())
				dbErr := db.StatusFailed(wid, fmt.Sprintf("colliding workflow name: %s; rename to something else", wid.String()), -1, n)
				if dbErr != nil {
					l.Error("failed to set workflow status to failed", "wid", wid, "err", dbErr)
				}
				observability.GetMetrics(ctx).RecordWorkflowTerminal(
					engineName(eng),
					"failure",
					string(FailureClassUser),
					string(FailureReasonWorkflowInvalid),
				)
				continue
			}

			wg.Go(func() {
				repoDID := w.RepoDID
				if repoDID == "" && pipeline != nil {
					repoDID = pipeline.RepoDid.String()
				}
				wl := l.With(
					"owner_did", w.OwnerDID,
					"repo_did", repoDID,
					"repo", db.RepoName(repoDID),
					"pipeline_id", pipelineId.String(),
					"workflow_id", wid.String(),
					"workflow", wid.Name,
				)

				if st, err := db.GetStatus(wid); err == nil && st.IsFinish() {
					wl.Info("skipping finished workflow", "wid", wid, "status", st)
					return
				}

				requestedResources := quota.Resources{}
				reporter, hasQuotaReporter := eng.(WorkflowQuotaReporter)
				if hasQuotaReporter {
					for resource, amount := range reporter.QuotaResources(&w) {
						if amount != 0 {
							requestedResources[resource] = amount
						}
					}
				}
				reqWorkflows := requestedResources[quota.ResourceWorkflows]
				reqMemoryMiB := requestedResources[quota.ResourceMemoryMiB]
				reqVCPUs := requestedResources[quota.ResourceVCPUs]
				reqDiskMiB := requestedResources[quota.ResourceDiskMiB]
				reqCacheBytes := requestedResources[quota.ResourceCacheStorageBytes]

				var err error
				var wfCtx context.Context = ctx
				engName := engineName(eng)

				var span trace.Span
				wfCtx, span = observability.Tracer().Start(wfCtx, "workflow.run", trace.WithAttributes(
					attribute.String(observability.WorkflowEngineKey, engName),
				))
				if span.IsRecording() {
					attrs := []attribute.KeyValue{
						attribute.String(observability.PipelineIDKey, pipelineId.String()),
						attribute.String(observability.WorkflowIDKey, wid.String()),
					}
					if w.OwnerDID != "" {
						attrs = append(attrs, attribute.String(observability.OwnerDIDKey, w.OwnerDID))
					}
					if repoDID != "" {
						attrs = append(attrs, attribute.String(observability.RepoDIDKey, repoDID))
					}
					attrs = append(attrs, observability.RequestedResourceAttrs(reqWorkflows, reqVCPUs, reqMemoryMiB, reqDiskMiB, reqCacheBytes)...)
					span.SetAttributes(attrs...)
				}
				defer span.End()

				_, remoteStatus := eng.(RemoteStatusEngine)
				metrics := observability.GetMetrics(ctx)
				if !remoteStatus {
					metrics.RecordWorkflowStart(engName)
				}
				startTime := time.Now()
				var runStart time.Time
				result := "success"
				remoteExecutionStarted := false

				defer func() {
					if err != nil {
						span.SetStatus(codes.Error, "workflow failed")
					} else if isCanceled(wfCtx) {
						span.SetStatus(codes.Error, "workflow cancelled")
					} else {
						span.SetStatus(codes.Ok, "success")
					}

					duration := time.Since(startTime)
					resStr := result
					if err != nil {
						resStr = workflowResult(wfCtx, err)
					} else if isCanceled(wfCtx) {
						resStr = "cancelled"
					}

					var cpuUsec, memoryCurrent, memoryPeak, swapCurrent, swapPeak, pidsCurrent, ioReadBytes, ioWriteBytes, ioReadOps, ioWriteOps, volumeAllocated uint64
					var cgroupAvailable, volumeAvailable bool
					if reporter, ok := eng.(WorkflowResourceUsageReporter); ok {
						if usage, ok := reporter.WorkflowResourceUsage(&w); ok {
							cpuUsec = usage.CPUUsec
							memoryCurrent = usage.MemoryCurrentBytes
							memoryPeak = usage.MemoryPeakBytes
							swapCurrent = usage.SwapCurrentBytes
							swapPeak = usage.SwapPeakBytes
							pidsCurrent = usage.PIDsCurrent
							ioReadBytes = usage.IOReadBytes
							ioWriteBytes = usage.IOWriteBytes
							ioReadOps = usage.IOReadOps
							ioWriteOps = usage.IOWriteOps
							volumeAllocated = usage.VolumeAllocatedBytes
							cgroupAvailable = usage.CgroupAvailable
							volumeAvailable = usage.VolumeAvailable
						}
					}

					waited := duration
					if !runStart.IsZero() {
						waited = runStart.Sub(startTime)
					}
					ran := duration - waited
					failureClass, failureReason := FailureAttribution(resStr, err)
					var errMsg string
					if err != nil {
						errMsg = truncateRunes(err.Error(), 300)
					}

					wl.InfoContext(wfCtx, "workflow finished",
						"result", resStr,
						"failure_class", failureClass,
						"failure_reason", failureReason,
						"error", errMsg,
						"duration_seconds", ran.Seconds(),
						"wait_seconds", waited.Seconds(),
						"requested_workflows", reqWorkflows,
						"requested_vcpus", reqVCPUs,
						"requested_memory_mib", reqMemoryMiB,
						"requested_disk_mib", reqDiskMiB,
						"requested_cache_bytes", reqCacheBytes,
						"actual_cpu_usec", cpuUsec,
						"actual_memory_current_bytes", memoryCurrent,
						"actual_memory_peak_bytes", memoryPeak,
						"actual_swap_current_bytes", swapCurrent,
						"actual_swap_peak_bytes", swapPeak,
						"actual_pids_current", pidsCurrent,
						"actual_io_read_bytes", ioReadBytes,
						"actual_io_write_bytes", ioWriteBytes,
						"actual_io_read_ops", ioReadOps,
						"actual_io_write_ops", ioWriteOps,
						"actual_volume_allocated_bytes", volumeAllocated,
						"actual_cgroup_available", cgroupAvailable,
						"actual_volume_available", volumeAvailable,
					)

					if span.IsRecording() {
						span.SetAttributes(
							observability.ActualResourceAttrs(
								cpuUsec, memoryCurrent, memoryPeak, swapCurrent, swapPeak, pidsCurrent,
								ioReadBytes, ioWriteBytes, ioReadOps, ioWriteOps, volumeAllocated,
								cgroupAvailable, volumeAvailable,
							)...,
						)
					}

					if remoteStatus {
						if !remoteExecutionStarted {
							metrics.RecordWorkflowTerminal(engName, resStr, failureClass, failureReason)
						}
						return
					}

					if observability.NotifyWorkflowTerminal(wfCtx, engName, resStr, failureClass, failureReason) {
						metrics.RecordWorkflowExecutionEnd(wfCtx, engName, resStr, ran)
					} else {
						metrics.RecordWorkflowEnd(wfCtx, engName, resStr, failureClass, failureReason, ran)
					}
				}()

				// opened only once the slot is taken, otherwise queued workflows would hold a file
				var wfLogger models.WorkflowLogger
				var closeLog func()
				defer func() {
					if closeLog != nil {
						archiveWorkflowLog(wl, stores, db, cfg.Server.LogDir, wid, repoDID)
					}
				}()

				var userCancel context.CancelCauseFunc
				wfCtx, userCancel = context.WithCancelCause(wfCtx)
				defer userCancel(nil)

				// allow wf context to be cancelled properly by manual cancel
				activeMu.Lock()
				activeCancels[wid] = userCancel
				activeMu.Unlock()
				defer func() {
					activeMu.Lock()
					delete(activeCancels, wid)
					activeMu.Unlock()
				}()

				slot := WorkflowSlot(NoopSlot{})
				var publishTerminalStatus func()
				var quotaLease quota.Lease
				destroyWorkflow := false
				slotAcquired := false
				setTerminalError := func(phase string, workflowErr error) {
					publishTerminalStatus = func() {
						writeWfError(db, n, wl, wfCtx, wid, phase, workflowErr)
					}
				}
				defer func() {
					if closeLog != nil {
						closeLog()
					}
					if publishTerminalStatus != nil {
						publishTerminalStatus()
					}
					if destroyWorkflow {
						if err := eng.DestroyWorkflow(ctx, wid); err != nil {
							wl.ErrorContext(wfCtx, "failed to destroy workflow", "wid", wid, "err", err)
						}
					}
					if slotAcquired {
						slot.Release()
					}
					if quotaLease != nil {
						quotaLease.Release()
					}
				}()

				quotaIdentity := quota.Identity{OwnerDID: w.OwnerDID, RepoDID: repoDID}
				quotaID := quota.WorkflowReservationID(w.RunID, quotaIdentity.OwnerDID, quotaIdentity.RepoDID, string(wid.PipelineId), wid.Name)

				if qm != nil && hasQuotaReporter {
					req := quota.ReserveRequest{
						ID:        quotaID,
						Kind:      quota.KindWorkflow,
						Key:       quotaID,
						Identity:  quotaIdentity,
						Resources: requestedResources,
					}
					quotaLease, err = qm.Acquire(wfCtx, req)
					if err != nil {
						if errors.Is(err, quota.ErrDenied) {
							err = ClassifiedFailure(FailureClassPolicy, FailureReasonQuotaDenied, err)
						}
						setTerminalError("acquiring quota", err)
						return
					}
				}

				wl.InfoContext(wfCtx, "waiting for slot", "wid", wid)

				if s, ok := eng.(WorkflowSlotter); ok {
					slot, err = s.AcquireWorkflowSlot(wfCtx, wid, &w, Wait)
					if err != nil {
						setTerminalError("waiting for slot", err)
						return
					}
					slotAcquired = true
					remoteExecutionStarted = true
				}

				// after the slot, or every queued run could pass this before any of them bills
				minutesBilled := qm != nil && quotaIdentity.Valid()
				if qm != nil && !minutesBilled {
					wl.WarnContext(wfCtx, "workflow has no valid owner and repo identity, so its ci minutes are neither checked nor billed")
				}
				if minutesBilled {
					if err = qm.CheckMinutes(wfCtx, quotaIdentity); err != nil {
						if errors.Is(err, quota.ErrDenied) {
							err = ClassifiedFailure(FailureClassPolicy, FailureReasonQuotaDenied, err)
						}
						// nothing ran remotely, so the terminal metric is still ours to record
						remoteExecutionStarted = false
						setTerminalError("checking ci minutes", err)
						return
					}
				}

				if p, ok := eng.(workflowLoggerProvider); ok {
					wfLogger = p.WorkflowLogger(wid)
				} else if fileLogger, err := models.NewFileWorkflowLogger(cfg.Server.LogDir, wid, secretValues, models.WithMaxLogBytes(cfg.Server.LogMaxBytes)); err != nil {
					wl.WarnContext(wfCtx, "failed to setup step logger; logs will not be persisted", "error", err)
					wfLogger = models.NullLogger{}
				} else {
					wl.InfoContext(wfCtx, "setup step logger; logs will be persisted", "logDir", cfg.Server.LogDir, "wid", wid)
					wfLogger = fileLogger
					var closeOnce sync.Once
					closeLog = func() {
						closeOnce.Do(func() {
							if err := fileLogger.Close(); err != nil {
								wl.ErrorContext(wfCtx, "failed to close workflow log", "wid", wid, "err", err)
							}
						})
					}
				}
				runStart = time.Now()

				// the timeout measures the run, not the queue wait above
				var timeoutCancel context.CancelFunc
				wfCtx, timeoutCancel = context.WithTimeout(wfCtx, workflowTimeout)
				defer timeoutCancel()

				runBudget := workflowTimeout
				if b, ok := slot.(RunBudgeter); ok && b.RunBudget() > 0 {
					runBudget = min(runBudget, b.RunBudget())
				}

				// the run's ID token is a credential for this workflow alone: it is
				// minted now, with the workflow holding its slot, and handed to the
				// steps and the log mask exactly like a vault secret
				stepSecrets := allSecrets
				if pipeline.MintIDToken != nil && isTrustedRepo && w.Audience != "" {
					token, mintErr := pipeline.MintIDToken(runBudget, w.Audience)
					if mintErr != nil {
						wl.ErrorContext(wfCtx, "failed to mint an id token", "wid", wid, "err", mintErr)
						setTerminalError("minting id token", mintErr)
						return
					}
					if token != "" {
						stepSecrets = slices.Concat(allSecrets, []secrets.UnlockedSecret{{
							Key:   models.IDTokenEnvVar,
							Value: token,
						}})
						if fileLogger, ok := wfLogger.(*models.FileWorkflowLogger); ok {
							fileLogger.AddSecretValues(token)
						}
					}
				}

				// attic being down costs the run its cache, not its result
				if pipeline.GrantNixCache != nil {
					grant, grantErr := pipeline.GrantNixCache(wfCtx, wid, runBudget)
					if grantErr != nil {
						wl.WarnContext(wfCtx, "running without a nix cache, the attic grant failed", "wid", wid, "err", grantErr)
					} else if grant != nil {
						w.NixCache = grant
					}
				}

				if !remoteStatus {
					err = db.StatusRunning(wid, n)
					if err != nil {
						wl.ErrorContext(wfCtx, "failed to set workflow status to running", "wid", wid, "err", err)
						return
					}
					if delay, ok, delayErr := db.WorkflowStartupDelay(wfCtx, wid); delayErr != nil {
						wl.WarnContext(wfCtx, "failed to measure workflow startup delay", "err", delayErr)
					} else if ok {
						metrics.RecordWorkflowStartupDelay(wfCtx, engName, "local", delay)
					}
				}

				// a spindle crash bills nothing, the defer never runs
				if minutesBilled {
					defer func() {
						minutes := quota.BillableMinutes(runStart, time.Now())
						if err := recordMinutes(context.WithoutCancel(wfCtx), qm, quotaID, quotaIdentity, minutes); err != nil {
							wl.ErrorContext(wfCtx, "failed to record ci minutes", "wid", wid, "minutes", minutes, "err", err)
						}
					}()
				}

				wl.InfoContext(wfCtx, "workflow started",
					"engine", engName,
					"requested_workflows", reqWorkflows,
					"requested_vcpus", reqVCPUs,
					"requested_memory_mib", reqMemoryMiB,
					"requested_disk_mib", reqDiskMiB,
					"requested_cache_bytes", reqCacheBytes,
				)

				err = eng.SetupWorkflow(wfCtx, wid, &w, wfLogger)
				if err != nil {
					err = ClassifiedFailure(FailureClassInfrastructure, FailureReasonSetupFailed, err)
					// a canceled setup may still have created resources
					destroyWorkflow = true
					if !remoteStatus {
						setTerminalError("setting up workflow", err)
					}
					return
				}
				destroyWorkflow = true

				var bindings []models.CacheBinding
				var trackedStore *trackedCacheStore
				if cacheEnabled && len(w.Caches) > 0 {
					bindings, err = cacheController.Plan(wfCtx, pipeline, &w)
					if err != nil {
						l.Warn("cache planning failed", "wid", wid, "err", err)
					} else {
						w.CacheBindings = bindings
						trackedStore = newTrackedCacheStore(cacheStore, cacheController, l, bindings)
						if cachesSupported {
							wfLogger.ControlWriter(CacheRestoreStepIdx, CacheRestoreStep, models.StepStatusStart).Write([]byte{0})
							// caches are an optimization, never a reason to fail the workflow
							if err := cacheRunner.RestoreCache(wfCtx, wid, &w, trackedStore, bindings, wfLogger); err != nil {
								l.Warn("cache restore failed", "wid", wid, "err", err)
							}
							wfLogger.ControlWriter(CacheRestoreStepIdx, CacheRestoreStep, models.StepStatusEnd).Write([]byte{0})
						} else if !remoteStatus {
							l.Warn("engine does not support caches, skipping restore", "wid", wid)
						}
					}
				}

				cleanupCaches := func() {
					if trackedStore != nil {
						trackedStore.cleanup(context.WithoutCancel(wfCtx))
						trackedStore = nil
					}
				}
				saveCaches := func(ctx context.Context, failed bool) {
					if trackedStore == nil || !cachesSupported {
						return
					}
					toSave := make([]models.CacheBinding, 0, len(bindings))
					for _, binding := range bindings {
						if binding.SaveKey != "" && binding.SaveOn(failed) {
							toSave = append(toSave, binding)
						}
					}
					if len(toSave) == 0 {
						return
					}
					wfLogger.ControlWriter(CacheSaveStepIdx, CacheSaveStep, models.StepStatusStart).Write([]byte{0})
					if err := cacheRunner.SaveCache(ctx, wid, &w, trackedStore, toSave, wfLogger); err != nil {
						l.Warn("cache save failed", "wid", wid, "err", err)
					}
					wfLogger.ControlWriter(CacheSaveStepIdx, CacheSaveStep, models.StepStatusEnd).Write([]byte{0})
				}
				for stepIdx, step := range w.Steps {
					if wfLogger != nil {
						wfLogger.
							ControlWriter(stepIdx, step, models.StepStatusStart).
							Write([]byte{0})
					}

					if !remoteStatus {
						metrics.RecordStepStart(engName)
					}
					stepStart := time.Now()

					var stepSpan trace.Span
					var stepCtx context.Context
					stepCtx, stepSpan = observability.Tracer().Start(wfCtx, "step.run")
					if stepSpan.IsRecording() {
						attrs := []attribute.KeyValue{
							attribute.String(observability.WorkflowEngineKey, engName),
							attribute.String(observability.PipelineIDKey, pipelineId.String()),
							attribute.String(observability.WorkflowIDKey, wid.String()),
							attribute.Int(observability.StepIndexKey, stepIdx),
							attribute.String(observability.StepNameKey, step.Name()),
						}
						if w.OwnerDID != "" {
							attrs = append(attrs, attribute.String(observability.OwnerDIDKey, w.OwnerDID))
						}
						if repoDID != "" {
							attrs = append(attrs, attribute.String(observability.RepoDIDKey, repoDID))
						}
						stepSpan.SetAttributes(attrs...)
					}

					err = eng.RunStep(stepCtx, wid, &w, stepIdx, stepSecrets, wfLogger)

					if wfLogger != nil {
						wfLogger.
							EndWriter(stepIdx, step, models.OutcomeOf(err)).
							Write([]byte{0})
					}

					stepResult := "success"
					if err != nil {
						stepResult = workflowResult(wfCtx, err)
						stepSpan.SetStatus(codes.Error, "step failed")
					} else {
						stepSpan.SetStatus(codes.Ok, "success")
					}
					stepSpan.End()

					if !remoteStatus {
						metrics.RecordStepEnd(stepCtx, engName, stepResult, time.Since(stepStart))
					}

					if err != nil {
						switch stepResult {
						case "failure":
							saveCaches(wfCtx, true)
						case "timeout":
							// wfCtx is at or past its deadline
							saveCtx, cancelSave := context.WithTimeout(context.WithoutCancel(wfCtx), CacheSaveTimeout)
							saveCaches(saveCtx, true)
							cancelSave()
						}
						cleanupCaches()
						if !remoteStatus {
							setTerminalError("running step", err)
						}
						return
					}
				}

				saveCaches(wfCtx, false)
				cleanupCaches()

				if isCanceled(wfCtx) {
					if !remoteStatus {
						setTerminalError("before success", nil)
					}
					return
				}

				if !remoteStatus {
					publishTerminalStatus = func() {
						if err := db.StatusSuccess(wid, n); err != nil {
							wl.ErrorContext(wfCtx, "failed to set workflow status to success", "wid", wid, "err", err)
						}
					}
				}
			})
		}
	}

	wg.Wait()
	l.Info("all workflows completed")
}

const recordMinutesAttempts = 3

// recording is idempotent by usage id, so a failed attempt can simply be tried again
func recordMinutes(ctx context.Context, qm *quota.Manager, usageID string, id quota.Identity, minutes int64) (err error) {
	for attempt := 1; attempt <= recordMinutesAttempts; attempt++ {
		if err = qm.RecordMinutes(ctx, usageID, id, minutes); err == nil {
			return nil
		}
		time.Sleep(time.Duration(attempt) * 100 * time.Millisecond)
	}
	return err
}

type logStores interface {
	PutFile(ctx context.Context, ref, sourcePath string) []artifactstore.PutResult
}

func archiveWorkflowLog(l *slog.Logger, stores *artifactstore.Stores, database *db.DB, logDir string, wid models.WorkflowId, repoDID string) {
	if stores == nil {
		return
	}
	archiveLogTo(l, stores, database, logDir, wid, repoDID)
}

func archiveLogTo(l *slog.Logger, stores logStores, database *db.DB, logDir string, wid models.WorkflowId, repoDID string) {
	logPath := models.LogFilePath(logDir, wid)
	file, err := os.Open(logPath)
	if err != nil {
		l.Error("open workflow log for archival", "wid", wid, "err", err)
		return
	}
	hash := sha256.New()
	if _, err := io.Copy(hash, file); err != nil {
		_ = file.Close()
		l.Error("hash workflow log", "wid", wid, "err", err)
		return
	}
	_ = file.Close()

	ref := wid.String() + ".log"
	uploadCtx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	defer cancel()
	stored, keep := false, false
	for _, result := range stores.PutFile(uploadCtx, ref, logPath) {
		if result.Err != nil {
			l.Error("archive workflow log", "wid", wid, "err", result.Err)
			keep = true
			continue
		}
		stored = true
		keep = keep || result.InPlace
	}
	if !stored {
		return
	}
	digest := "sha256:" + hex.EncodeToString(hash.Sum(nil))
	if err := database.SaveArtifactRef(wid.String(), repoDID, wid, ref, digest); err != nil {
		l.Error("save workflow log artifact", "wid", wid, "err", err)
		return
	}
	// a store that failed or holds the file in place leaves the log as the only copy
	if keep {
		return
	}
	if err := os.Remove(logPath); err != nil {
		l.Warn("remove archived workflow log", "wid", wid, "err", err)
	}
}