package spindle

import (
	"context"
	_ "embed"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"maps"
	"net/http"
	"net/url"
	"path/filepath"
	"regexp"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/bluesky-social/indigo/atproto/syntax"
	"github.com/go-chi/chi/v5"
	"github.com/hashicorp/go-version"
	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/codes"
	"golang.org/x/sync/singleflight"
	"tangled.org/core/api/org_tangled"
	"tangled.org/core/gitutil"
	"tangled.org/core/idresolver"
	"tangled.org/core/jetstream"
	"tangled.org/core/log"
	"tangled.org/core/notifier"
	"tangled.org/core/rbac"
	"tangled.org/core/spindle/artifactstore"
	"tangled.org/core/spindle/config"
	"tangled.org/core/spindle/db"
	"tangled.org/core/spindle/engine"
	"tangled.org/core/spindle/engines/dummy"
	"tangled.org/core/spindle/engines/nixery"
	"tangled.org/core/spindle/identity"
	"tangled.org/core/spindle/knotclient"
	"tangled.org/core/spindle/mill"
	"tangled.org/core/spindle/mill/executor"
	"tangled.org/core/spindle/models"
	"tangled.org/core/spindle/netguard"
	"tangled.org/core/spindle/nixcache"
	"tangled.org/core/spindle/observability"
	"tangled.org/core/spindle/oidc"
	"tangled.org/core/spindle/pemkey"
	pipelinecodec "tangled.org/core/spindle/pipeline"
	"tangled.org/core/spindle/quota"
	"tangled.org/core/spindle/secrets"
	"tangled.org/core/spindle/storage"
	"tangled.org/core/spindle/webhook"
	"tangled.org/core/spindle/xrpc"
	"tangled.org/core/tid"
	"tangled.org/core/workflow"
	"tangled.org/core/xrpc/serviceauth"
)

//go:embed motd
var defaultMotd []byte

const (
	rbacDomain              = "thisserver"
	executorShutdownTimeout = 8 * time.Minute
	serverShutdownTimeout   = 10 * time.Second
	dequeueRetryDelay       = time.Second
	// attic checks the token per request, and the last upload can land after the budget
	nixCacheTokenGrace = 5 * time.Minute

	// sparse-checkout wants a leading slash to anchor the pattern at the root
	sparseWorkflowDir = "/" + workflow.WorkflowDir
)

type executorClient interface {
	Connect(context.Context)
	Drain(context.Context) error
	RegisterMetrics(*observability.Metrics)
	SetQuotaClient(*executor.QuotaClient)
}

type Spindle struct {
	jc          *jetstream.JetstreamClient
	backfill    *backfill
	db          *db.DB
	e           *rbac.Enforcer
	l           *slog.Logger
	n           *notifier.Notifier
	wh          *webhook.Service
	metrics     *observability.Metrics
	engs        map[string]models.Engine
	debugKeys   debugKeyCache
	jobWake     chan struct{}
	jobWorkers  sync.WaitGroup
	jobsMu      sync.Mutex
	jobsByOwner map[string]int
	scheduler   *pipelineScheduler
	cfg         *config.Config
	res         *idresolver.Resolver
	vault       secrets.Manager
	cache       storage.Storage
	cacheCtl    engine.CacheController
	motd        []byte
	motdMu      sync.RWMutex
	rootCtx     context.Context
	rootCancel  context.CancelFunc
	store       artifactstore.Store
	stores      *artifactstore.Stores
	reader      artifactstore.Reader
	qm          *quota.Manager
	// set only when this spindle hosts the mill or joins one as an executor
	mill      *mill.Mill
	oidc      *oidc.Issuer
	nixCache  *nixcache.Issuer
	identity  http.HandlerFunc
	exec      executorClient
	repoLocks stripedLocks
	turns     turnClock
	retries   repoRetries
	knots     *knotclient.Client

	pushGroup           singleflight.Group
	notifySlots         notifySlots
	notifyWork          sync.WaitGroup
	notifyPriorityOnce  sync.Once
	notifyPriorityHosts map[string]bool
	notifyAdmission     notifyAdmission
	workflowDefs        workflowDefCache
}

func newCacheStore(ctx context.Context, cfg *config.Config) (storage.Storage, error) {
	switch cfg.Cache.Backend {
	case "":
		return nil, nil
	case "disk":
		dir := cfg.Cache.DiskDir
		if dir == "" {
			dir = filepath.Join(filepath.Dir(cfg.Server.DBPath), "cache")
		}
		return storage.NewDisk(dir)
	case "s3":
		return storage.NewUnversionedS3(ctx, cfg.Cache.S3Bucket, cfg.Cache.S3Prefix)
	default:
		return nil, fmt.Errorf("storage: unknown backend %q", cfg.Cache.Backend)
	}
}

// New creates a new Spindle server with the provided configuration and engines.
func New(ctx context.Context, cfg *config.Config, d *db.DB, engines map[string]models.Engine, qm *quota.Manager) (*Spindle, error) {
	logger := log.FromContext(ctx)
	metrics := observability.GetMetrics(ctx)
	if metrics == nil {
		metrics = observability.NewMetrics()
		ctx = observability.WithMetrics(ctx, metrics)
	}
	n := notifier.New()
	lifecycleCtx, cancelLifecycle := context.WithCancel(context.WithoutCancel(ctx))
	keepLifecycle := false
	defer func() {
		if !keepLifecycle {
			cancelLifecycle()
		}
	}()

	if err := cleanupOrphanRepos(ctx, d, logger); err != nil {
		return nil, fmt.Errorf("failed to run startup cleanup: %w", err)
	}

	cacheStore, err := newCacheStore(ctx, cfg)
	if err != nil {
		return nil, fmt.Errorf("failed to setup cache storage: %w", err)
	}
	if cacheStore != nil {
		logger.Info("cache storage enabled", "backend", cfg.Cache.Backend, "storeID", cfg.Cache.StoreID)
	}

	var issuer *oidc.Issuer
	if cfg.Role != config.RoleExecutor && cfg.OIDC.KeyPath != "" {
		issuer, err = oidc.New(oidc.Options{
			Issuer:  cfg.OIDC.Issuer,
			KeyPath: cfg.OIDC.KeyPath,
			KeyID:   cfg.OIDC.KeyID,
		})
		if err != nil {
			return nil, fmt.Errorf("failed to setup OIDC issuer: %w", err)
		}
		logger.Info("minting CI ID tokens", "issuer", cfg.OIDC.Issuer, "kid", issuer.KeyID())
	}

	var nixIssuer *nixcache.Issuer
	if cfg.Role != config.RoleExecutor && cfg.NixCache.AtticURL != "" {
		key, err := pemkey.LoadRSA(cfg.NixCache.AtticKeyPath, "attic key")
		if err != nil {
			return nil, fmt.Errorf("setting up attic cache tokens: %w", err)
		}
		nixIssuer = &nixcache.Issuer{
			Client:         nixcache.Client{BaseURL: cfg.NixCache.AtticURL},
			Key:            key,
			RunUploadBytes: cfg.NixCache.RunUploadMiB << 20,
		}
		logger.Info("minting attic cache tokens", "url", cfg.NixCache.AtticURL)
	}

	var id http.HandlerFunc
	if cfg.Role != config.RoleExecutor && cfg.Server.KeyPath != "" {
		id, err = identity.Load(cfg.Server.Hostname, cfg.Server.KeyPath)
		if err != nil {
			return nil, fmt.Errorf("failed to load spindle identity: %w", err)
		}
		logger.Info("serving did document", "did", cfg.Server.Did())
	}

	spindle := &Spindle{
		db:         d,
		l:          logger,
		n:          &n,
		wh:         webhook.New(d, cfg.Server.Dev, pauseGate(d, &cfg.Server)),
		metrics:    metrics,
		engs:       engines,
		cfg:        cfg,
		cache:      cacheStore,
		motd:       defaultMotd,
		rootCtx:    lifecycleCtx,
		rootCancel: cancelLifecycle,
		jobWake:    make(chan struct{}, 1),
		qm:         qm,
		oidc:       issuer,
		nixCache:   nixIssuer,
		identity:   id,
	}
	metrics.AttachDB(ctx, d)
	diskFallback := ""
	if cfg.Role == config.RoleStandalone {
		diskFallback = cfg.Server.LogDir
		if cfg.ArtifactStores.Disk.Dir == "" {
			logger.Warn("using SPINDLE_SERVER_LOG_DIR as the implicit disk artifact store; configure SPINDLE_ARTIFACT_STORES_DISK_DIR explicitly")
		}
	}
	stores, err := artifactstore.NewStores(cfg.ArtifactStores, diskFallback, cfg.LegacyS3.LogBucket)
	if err != nil {
		return nil, fmt.Errorf("failed to setup artifact stores: %w", err)
	}
	spindle.stores = stores
	if cfg.LegacyS3.LogBucket != "" {
		logger.Warn("SPINDLE_S3_LOG_BUCKET is deprecated; use SPINDLE_ARTIFACT_STORES_S3_BUCKET")
	}
	if cfg.Role != config.RoleExecutor {
		engine.StartCachePruner(ctx, logger, d, cacheStore, cfg.Cache.Retention, cfg.Cache.PruneInterval, cfg.Cache.MaxBytes)
	}
	if cfg.Role == config.RoleStandalone {
		spindle.reader = stores
	} else {
		name := cfg.Mill.ArtifactStore
		if name == "" {
			names := stores.Names()
			if len(names) != 1 {
				return nil, fmt.Errorf("%s requires SPINDLE_MILL_ARTIFACT_STORE when %d artifact stores are configured", cfg.Role, len(names))
			}
			name = names[0]
			logger.Warn("SPINDLE_MILL_ARTIFACT_STORE is not set; inferred the only configured store", "store", name)
		}
		store, ok := stores.Store(name)
		if !ok {
			return nil, fmt.Errorf("SPINDLE_MILL_ARTIFACT_STORE=%q is not configured", name)
		}
		spindle.store = store
		spindle.reader = store
	}
	if cfg.Role == config.RoleExecutor {
		keepLifecycle = true
		return spindle, nil
	}

	e, err := rbac.NewEnforcer(cfg.Server.DBPath)
	if err != nil {
		return nil, fmt.Errorf("failed to setup rbac enforcer: %w", err)
	}
	e.E.EnableAutoSave(true)
	spindle.e = e

	switch cfg.Server.Secrets.Provider {
	case "openbao":
		if cfg.Server.Secrets.OpenBao.ProxyAddr == "" {
			return nil, fmt.Errorf("openbao proxy address is required when using openbao secrets provider")
		}
		spindle.vault, err = secrets.NewOpenBaoManager(
			cfg.Server.Secrets.OpenBao.ProxyAddr,
			logger,
			secrets.WithMountPath(cfg.Server.Secrets.OpenBao.Mount),
		)
		if err != nil {
			return nil, fmt.Errorf("failed to setup openbao secrets provider: %w", err)
		}
		logger.Info("using openbao secrets provider", "proxy_address", cfg.Server.Secrets.OpenBao.ProxyAddr, "mount", cfg.Server.Secrets.OpenBao.Mount)
	case "sqlite", "":
		spindle.vault, err = secrets.NewSQLiteManager(cfg.Server.DBPath, secrets.WithTableName("secrets"))
		if err != nil {
			return nil, fmt.Errorf("failed to setup sqlite secrets provider: %w", err)
		}
		logger.Info("using sqlite secrets provider", "path", cfg.Server.DBPath)
	default:
		return nil, fmt.Errorf("unknown secrets provider: %s", cfg.Server.Secrets.Provider)
	}

	collections := []string{
		org_tangled.RepoDeclarationNSID,
		org_tangled.RepoManifestNSID,
		org_tangled.GraphCollaboratorNSID,
		org_tangled.TrackTicketNSID,
		org_tangled.ReviewPatchNSID,
	}
	// the trailing false is logDids, on would log a line for every did added
	jc, err := jetstream.NewJetstreamClient(cfg.Server.JetstreamEndpoint, collections, nil, log.SubLogger(logger, "jetstream"), d, true, false)
	if err != nil {
		return nil, fmt.Errorf("failed to setup jetstream client: %w", err)
	}
	spindle.jc = jc
	jc.AddDid(cfg.Server.Owner)
	for _, admin := range cfg.Server.Admins {
		jc.AddDid(admin)
	}
	// repo dids aren't known up front, so these skip the did filter
	jc.ExemptCollection(org_tangled.RepoManifestNSID)
	jc.ExemptCollection(org_tangled.GraphCollaboratorNSID)
	jc.ExemptCollection(org_tangled.TrackTicketNSID)
	jc.ExemptCollection(org_tangled.ReviewPatchNSID)

	allowlisted, err := db.ListAllowlist(d)
	if err != nil {
		return nil, fmt.Errorf("failed to load the allow-list: %w", err)
	}
	for _, did := range allowlisted {
		jc.AddDid(did.String())
	}

	knownRepos, err := d.AllRepos()
	if err != nil {
		return nil, fmt.Errorf("failed to get known repos: %w", err)
	}
	for _, r := range knownRepos {
		if r.Owner != "" {
			jc.AddDid(r.Owner.String())
		}
	}

	spindle.res = idresolver.DefaultResolver(cfg.Server.PlcUrl)
	guarded := knotclient.HTTPClient(cfg.Server.Dev)
	spindle.knots = knotclient.New(spindle.res.Directory(), guarded, cfg.Server.Dev)
	spindle.scheduler = newPipelineScheduler(spindle)

	err = e.AddSpindle(rbacDomain)
	if err != nil {
		return nil, fmt.Errorf("failed to set rbac domain: %w", err)
	}
	err = spindle.configureOwner()
	if err != nil {
		return nil, err
	}
	logger.Info("owner set", "did", cfg.Server.Owner)

	spindle.backfill = newBackfill(spindle, log.SubLogger(logger, "backfill"), cfg.Server.Hydrant.Dir)
	jc.SetGapHandler(func(context.Context) { spindle.backfill.CatchUpLater() })
	err = jc.StartJetstream(lifecycleCtx, spindle.ingest())
	if err != nil {
		return nil, fmt.Errorf("failed to start jetstream consumer: %w", err)
	}

	keepLifecycle = true
	return spindle, nil
}
func (s *Spindle) DB() *db.DB {
	return s.db
}

func (s *Spindle) Engines() map[string]models.Engine {
	return s.engs
}

func (s *Spindle) Vault() secrets.Manager {
	return s.vault
}

func (s *Spindle) Notifier() *notifier.Notifier {
	return s.n
}

func (s *Spindle) Enforcer() *rbac.Enforcer {
	return s.e
}

func (s *Spindle) SetMotdContent(content []byte) {
	s.motdMu.Lock()
	defer s.motdMu.Unlock()
	s.motd = content
}

func (s *Spindle) GetMotdContent() []byte {
	s.motdMu.RLock()
	defer s.motdMu.RUnlock()
	return s.motd
}

// runs the server. blocks
func (s *Spindle) Start(ctx context.Context) error {
	runCtx := s.rootCtx
	cancelRun := s.rootCancel
	if runCtx == nil || cancelRun == nil {
		runCtx, cancelRun = context.WithCancel(context.WithoutCancel(ctx))
		s.rootCtx = runCtx
		s.rootCancel = cancelRun
	}
	defer cancelRun()

	_, err := observability.StartMetricsServer(runCtx, s.cfg.Server.MetricsListenAddr, s.l, s.metrics.Registry())
	if err != nil {
		return fmt.Errorf("starting metrics listener: %w", err)
	}
	if s.scheduler != nil {
		s.scheduler.Start(runCtx)
	}

	// an executor dials out to its mill and takes work from it
	var execDone chan struct{}
	if s.exec != nil {
		execDone = make(chan struct{})
		go func() {
			defer close(execDone)
			s.exec.Connect(runCtx)
		}()
	}
	waitForExecutor := func(waitCtx context.Context) error {
		if execDone == nil {
			return nil
		}
		select {
		case <-execDone:
			return nil
		case <-waitCtx.Done():
			return waitCtx.Err()
		}
	}

	if s.mill != nil && s.cfg.Mill.JumpListenAddr != "" {
		go s.mill.ServeJump(runCtx, s.cfg.Mill.JumpListenAddr, s.cfg.Mill.JumpHostKeyPath, s.cfg.Mill.DebugExecutorPort, s.cfg.Mill.MaxJumpConnections)
	}

	if stopper, ok := s.vault.(secrets.Stopper); ok {
		defer stopper.Stop()
	}

	if s.cfg.Role != config.RoleExecutor {
		s.resumeWipes(ctx)

		failed, err := s.db.FailOrphanedWorkflows(s.n)
		if err != nil {
			return fmt.Errorf("failing orphaned workflows: %w", err)
		}
		if failed > 0 {
			s.l.Warn("failed workflows a restart left behind", "count", failed)
		}

		owners, err := s.backfillOwners()
		if err != nil {
			return err
		}
		s.backfill.Track(owners...)
		if err := s.backfill.Resume(); err != nil {
			return err
		}
		var repoSync sync.WaitGroup
		repoSync.Add(2)
		go func() {
			defer repoSync.Done()
			s.retryFailedRepos(runCtx)
		}()
		go func() {
			defer repoSync.Done()
			s.backfill.run(runCtx)
		}()
		// hydrant has to be shut down before the process exits, or its db is
		// left for the next start to recover
		defer func() {
			cancelRun()
			repoSync.Wait()
		}()
	}

	// start workers after resuming wipes so banned jobs cannot run
	workersCtx, cancelWorkers := context.WithCancel(runCtx)
	s.StartJobWorkers(workersCtx)
	defer func() {
		cancelWorkers()
		done := make(chan struct{})
		go func() {
			s.jobWorkers.Wait()
			close(done)
		}()
		select {
		case <-done:
		case <-time.After(5 * time.Second):
			s.l.Warn("timed out waiting for job workers to stop")
		}
	}()

	server := newHTTPServer(s.cfg.Server.ListenAddr, s.Router())
	serverErr := make(chan error, 1)
	go func() {
		s.l.Info("starting spindle server", "address", s.cfg.Server.ListenAddr)
		serverErr <- server.ListenAndServe()
	}()

	var listenErr error
	select {
	case listenErr = <-serverErr:
	case <-ctx.Done():
	}

	if s.exec != nil {
		drainCtx, cancelDrain := context.WithTimeout(context.Background(), s.cfg.Mill.DrainTimeout)
		if err := s.exec.Drain(drainCtx); err != nil {
			s.l.Warn("executor drain did not complete", "err", err)
		}
		cancelDrain()
	}

	if s.cfg.Role != config.RoleExecutor {
		webhookCtx, cancelWebhooks := context.WithTimeout(context.Background(), serverShutdownTimeout)
		s.wh.Close(webhookCtx)
		cancelWebhooks()
	}

	cancelRun()
	waitCtx, cancelWait := context.WithTimeout(context.Background(), executorShutdownTimeout)
	if err := waitForExecutor(waitCtx); err != nil {
		s.l.Warn("executor shutdown did not complete", "err", err)
	}
	cancelWait()

	// the final cursor save has to land before the db closes
	if s.jc != nil {
		select {
		case <-s.jc.Done():
		case <-time.After(serverShutdownTimeout):
			s.l.Warn("jetstream did not save its cursor before shutdown")
		}
	}

	shutdownCtx, cancelShutdown := context.WithTimeout(context.Background(), serverShutdownTimeout)
	shutdownErr := server.Shutdown(shutdownCtx)
	cancelShutdown()
	// give notify work a moment before the run ends, since it can outlast the handler
	notifyDone := make(chan struct{})
	go func() {
		s.notifyWork.Wait()
		close(notifyDone)
	}()
	select {
	case <-notifyDone:
	case <-time.After(5 * time.Second):
		s.l.Warn("timed out waiting for notify work to stop")
	}
	if listenErr == nil {
		listenErr = <-serverErr
	}
	if errors.Is(listenErr, http.ErrServerClosed) {
		listenErr = nil
	}
	if shutdownErr != nil {
		shutdownErr = fmt.Errorf("shutting down spindle server: %w", shutdownErr)
	}
	return errors.Join(listenErr, shutdownErr)
}

// the first round erases hydrant's copy of anyone missing, so a load error fails the whole list
func (s *Spindle) backfillOwners() ([]syntax.DID, error) {
	repos, err := s.db.AllRepos()
	if err != nil {
		return nil, fmt.Errorf("loading repo owners to backfill: %w", err)
	}
	allowed, err := db.ListAllowlist(s.db)
	if err != nil {
		return nil, fmt.Errorf("loading the allow-list to backfill: %w", err)
	}
	seen := make(map[syntax.DID]struct{}, len(repos)+len(allowed)+1)
	dids := make([]syntax.DID, 0, len(repos)+len(allowed)+1)
	add := func(did syntax.DID) {
		if did == "" {
			return
		}
		if _, ok := seen[did]; ok {
			return
		}
		seen[did] = struct{}{}
		dids = append(dids, did)
	}

	for _, admin := range append([]string{s.cfg.Server.Owner}, s.cfg.Server.Admins...) {
		did, err := syntax.ParseDID(admin)
		if err != nil {
			s.l.Warn("backfill: invalid configured admin", "admin", admin, "err", err)
			continue
		}
		add(did)
	}
	for _, member := range allowed {
		add(member)
	}
	for _, repo := range repos {
		add(repo.Owner)
	}
	sort.Slice(dids, func(i, j int) bool { return dids[i] < dids[j] })
	return dids, nil
}

var errMicrovmUnsupported = errors.New("microvm engine is only supported on Linux")

func localEngines(ctx context.Context, cfg *config.Config, d *db.DB, qStore quota.ReservationStore, logger *slog.Logger) (map[string]models.Engine, error) {
	nixeryEng, err := nixery.New(ctx, cfg)
	if err != nil {
		return nil, err
	}
	engines := map[string]models.Engine{
		"nixery": nixeryEng,
		"dummy":  dummy.New(logger),
	}
	microvmEng, err := newMicrovmEngine(ctx, cfg, d, qStore)
	switch {
	case errors.Is(err, errMicrovmUnsupported):
		logger.Warn("microvm engine unavailable, workflows that use it will fail", "error", err)
	case err != nil:
		return nil, err
	default:
		engines["microvm"] = microvmEng
	}
	return engines, nil
}

func Run(ctx context.Context) error {
	cfg, err := config.Load(ctx)
	if err != nil {
		return fmt.Errorf("failed to load config: %w", err)
	}

	shutdownTracing, err := observability.InitTracing(ctx, cfg.Tracing)
	if err != nil {
		return fmt.Errorf("failed to initialize tracing: %w", err)
	}
	defer func() {
		shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		if err := shutdownTracing(shutdownCtx); err != nil {
			log.FromContext(ctx).Error("failed to shut down tracing", "err", err)
		}
	}()

	otelHandler, shutdownLogging, err := observability.InitLogging(ctx, cfg.Logging)
	if err != nil {
		return fmt.Errorf("failed to initialize logging: %w", err)
	}
	defer func() {
		shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		if err := shutdownLogging(shutdownCtx); err != nil {
			log.FromContext(ctx).Error("failed to shut down logging", "err", err)
		}
	}()

	currentLogger := log.FromContext(ctx)
	combinedHandler := log.NewFanoutHandler(currentLogger.Handler(), otelHandler)
	combinedLogger := slog.New(combinedHandler)
	slog.SetDefault(combinedLogger)
	ctx = log.IntoContext(ctx, combinedLogger)

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

	if err := ensureGitVersion(); err != nil {
		return fmt.Errorf("ensuring git version: %w", err)
	}

	releaseInstance, err := lockInstance(cfg.Server.DBPath)
	if err != nil {
		return err
	}
	defer releaseInstance()

	d, err := db.Make(ctx, cfg.Server.DBPath)
	if err != nil {
		return fmt.Errorf("failed to setup db: %w", err)
	}
	if err := d.MigratePipelineLogFiles(cfg.Server.LogDir); err != nil {
		return fmt.Errorf("failed to migrate pipeline log files: %w", err)
	}

	var qs *db.QuotaStore
	var qm *quota.Manager
	var publishing []db.NixPublishing
	if cfg.Role != config.RoleExecutor {
		defaults, err := cfg.ToDefaults()
		if err != nil {
			return fmt.Errorf("invalid quota configuration: %w", err)
		}
		for scope, resources := range defaults {
			for resource, limit := range resources {
				metrics.SetQuotaDefaultLimit(string(scope), string(resource), limit)
			}
		}
		qs = db.NewQuotaStore(d, defaults)
		qm = quota.NewManager(qs, 50*time.Millisecond, metrics.QuotaObserver())
		defer qm.Close()

		if cfg.Role != config.RoleMill {
			publishing, err = qs.Recover(ctx, nil)
			if err != nil {
				return fmt.Errorf("failed to recover quota reservations: %w", err)
			}
		}
	}

	logger := log.FromContext(ctx)

	if cfg.SharesNixUploadWithOpenAdmission() {
		logger.Warn("SPINDLE_NIX_CACHE_UPLOAD_URL is shared by every repository and admission is open; any workflow can poison it")
	}

	var engines map[string]models.Engine
	var m *mill.Mill
	var quotaClient *executor.QuotaClient

	if cfg.Role == config.RoleMill {
		// on a mill host, engines place jobs on executors instead of running
		// them. all names share one Mill
		m = mill.New(log.SubLogger(logger, "mill"), mill.Config{
			LogDir:                cfg.Server.LogDir,
			MaxPending:            cfg.Mill.MaxPending,
			ReconnectGrace:        cfg.Mill.ReconnectGrace,
			CancelAckTimeout:      cfg.Mill.CancelAckTimeout,
			CancelTeardownTimeout: cfg.Mill.CancelTeardownTimeout,
			CacheStoreID:          cfg.Cache.StoreID,
			CacheQuota:            qs,
			CacheMaxBytes:         cfg.Cache.MaxBytes,
			NixCaches:             cfg.NixCache.AtticURL != "",
		})
		engines = map[string]models.Engine{
			"nixery":  mill.NewEngine("nixery", m),
			"microvm": mill.NewEngine("microvm", m),
			"dummy":   mill.NewEngine("dummy", m),
		}
	} else {
		// standalone and executor both run real engines locally
		var engineQStore quota.ReservationStore
		if cfg.Role == config.RoleExecutor {
			quotaClient = executor.NewQuotaClient()
		} else {
			engineQStore = qs
		}
		engines, err = localEngines(ctx, cfg, d, engineQStore, logger)
		if err != nil {
			return err
		}
	}

	s, err := New(ctx, cfg, d, engines, qm)
	if err != nil {
		return err
	}
	// an executor's workflows carry a check that asks the mill instead
	if cfg.Role != config.RoleExecutor {
		attachDebugAuthorizer(engines, s)
	}
	if s.nixCache != nil {
		s.nixCache.Limits = qs
		s.nixCache.Grants = qs
	}
	if s.cache != nil && cfg.Role != config.RoleExecutor {
		s.cacheCtl = engine.NewLocalCacheController(d, qs, s.cache, cfg.Server.RepoDir, cfg.Cache.MaxEntriesPerOwner, cfg.Cache.MaxBytes, logger)
	}

	if cfg.Role == config.RoleMill || cfg.Role == config.RoleStandalone {
		// flattens the store snapshot into bounded label tuples, the collector
		// drops anything outside the bounded scope, resource and status sets
		metrics.SetQuotaLoader(func() (observability.QuotaSnapshot, error) {
			snap, err := qs.MetricsSnapshot(context.Background())
			if err != nil {
				return observability.QuotaSnapshot{}, err
			}
			var out observability.QuotaSnapshot
			for scope, resources := range snap.Usage {
				for resource, used := range resources {
					out.Usage = append(out.Usage, observability.QuotaUsage{
						Scope:    string(scope),
						Resource: resource,
						Used:     used,
					})
				}
			}
			for scope, resources := range snap.Subjects {
				for resource, statuses := range resources {
					for status, count := range statuses {
						out.Subjects = append(out.Subjects, observability.QuotaSubjectCount{
							Scope:    string(scope),
							Resource: string(resource),
							Status:   status,
							Count:    count,
						})
					}
				}
			}
			return out, nil
		})
	}
	if m != nil {
		// the engines built above hold the mill, but the mill's db and
		// notifier only exist after New, so attach them here
		m.Attach(s.DB(), s.Notifier(), qm)
		m.AttachCache(s.cache)
		s.mill = m
		s.mill.SetDebugKeyChecker(s)
		s.mill.RegisterMetrics(s.metrics)
		if err := m.RestoreState(); err != nil {
			return fmt.Errorf("restoring mill state: %w", err)
		}
		liveIDs := m.LiveQuotaReservationIDs()
		for i := range liveIDs {
			liveIDs[i] = quota.StorageReservationID(liveIDs[i])
		}
		publishing, err = qs.Recover(ctx, liveIDs)
		if err != nil {
			return fmt.Errorf("failed to recover mill quota reservations: %w", err)
		}
	}
	var cacheKinds []quota.CacheKind
	if s.cacheCtl != nil {
		cacheKinds = append(cacheKinds, engine.CacheEntries(d, s.cache))
	}
	if s.nixCache != nil {
		dropped, bytes, err := qs.DropUnindexedNixCharges(ctx)
		if err != nil {
			return fmt.Errorf("dropping upload url nix cache charges: %w", err)
		}
		if dropped > 0 {
			logger.Info("dropped nix cache charges left from the upload url", "count", dropped, "bytes", bytes)
		}
		pruner := &nixcache.Pruner{
			Index:        qs,
			Attic:        s.nixCache,
			MaxBytes:     cfg.NixCache.MaxMiB << 20,
			Retention:    cfg.NixCache.Retention,
			Grace:        nixCacheTokenGrace,
			Grants:       qs,
			Publishing:   publishing,
			Reservations: qm,
			Log:          logger,
		}
		pruner.Start(ctx, cfg.Cache.PruneInterval)
		cacheKinds = append(cacheKinds, pruner)
	} else {
		// without attic nothing else settles these uploads
		for _, row := range publishing {
			if err := qm.Release(ctx, row.ReservationID); err != nil {
				return fmt.Errorf("releasing interrupted attic upload %s: %w", row.ReservationID, err)
			}
		}
	}
	if qs != nil {
		evictor := &quota.Evictor{Usage: qs, Limits: qs, Kinds: cacheKinds, Log: logger}
		evictor.Start(ctx, cfg.Cache.PruneInterval)
	}
	if cfg.Role == config.RoleExecutor {
		s.exec, err = executor.New(cfg, engines, s.DB(), s.Notifier(), log.SubLogger(logger, "executor"), s.store, s.cache)
		if err != nil {
			return err
		}
		s.exec.SetQuotaClient(quotaClient)
		s.exec.RegisterMetrics(s.metrics)
	}

	return s.Start(ctx)
}

// no write timeout, log streams run for hours and would get cut off
func newHTTPServer(addr string, handler http.Handler) *http.Server {
	return &http.Server{
		Addr:              addr,
		Handler:           handler,
		ReadHeaderTimeout: 10 * time.Second,
		ReadTimeout:       30 * time.Second,
		IdleTimeout:       2 * time.Minute,
	}
}

func (s *Spindle) Router() http.Handler {
	mux := chi.NewRouter()
	mux.Use(observability.HTTPMiddleware(s.metrics))
	if s.cfg.Tracing.Endpoint != "" {
		mux.Use(observability.OTelRouteMiddleware)
	}

	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
		w.Write(s.GetMotdContent())
	})
	mux.Get("/_health", func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/json")
		_, _ = io.WriteString(w, "{\"status\":\"ok\"}\n")
	})
	if s.cfg.Role != config.RoleExecutor {
		if s.mill != nil {
			mux.HandleFunc("/mill", s.mill.HandleExecutorConn)
		}

		mux.Mount("/xrpc", s.XrpcRouter())
	}

	if s.identity != nil {
		mux.Get(identity.DocumentPath, s.identity)
	}

	if s.oidc != nil {
		mux.Get(oidc.ConfigurationPath, s.oidc.ServeConfiguration)
		mux.Get(oidc.JWKSPath, s.oidc.ServeJWKS)
	}

	if s.cfg.Tracing.Endpoint == "" {
		return mux
	}
	return otelhttp.NewHandler(
		mux,
		"HTTP",
		otelhttp.WithSpanNameFormatter(func(string, *http.Request) string { return "HTTP" }),
	)
}

func (s *Spindle) XrpcRouter() http.Handler {
	serviceAuth := serviceauth.NewServiceAuth(s.l, s.res.Directory(), s.cfg.Server.Did().String())

	l := log.SubLogger(s.l, "xrpc")

	x := xrpc.Xrpc{
		Logger:         l,
		Db:             s.db,
		Enforcer:       s.e,
		Engines:        s.engs,
		Config:         s.cfg,
		ArtifactReader: s.reader,
		Resolver:       s.res,
		Vault:          s.vault,
		Notifier:       s.Notifier(),
		Webhooks:       s.wh,
		ServiceAuth:    serviceAuth,
		Trigger:        s,
		QuotaStore:     db.NewQuotaStore(s.db, quota.Defaults{}),
		Wiper:          s,
		Backfill:       s.backfill,
		Knots:          s.knots,
		Events:         s,
		Repos:          s,
	}

	return x.Router()
}

func (s *Spindle) buildTriggerRepo(ctx context.Context, repo *db.Repo) (*workflow.TriggerRepo, error) {
	rkey := string(repo.Rkey)
	repoDid := repo.RepoDid.String()
	return s.buildTriggerRepoFrom(ctx, repo.Knot, repo.Owner.String(), rkey, repoDid), nil
}

func (s *Spindle) buildTriggerRepoFrom(ctx context.Context, knot, did, rkey, repoDid string) *workflow.TriggerRepo {
	// an unreadable manifest only costs the trigger its default branch
	defaultBranch, _ := s.defaultBranch(ctx, knotclient.Knot{Host: knot}, syntax.DID(repoDid))

	var rkeyPtr *string
	if rkey != "" {
		rkeyPtr = &rkey
	}
	return &workflow.TriggerRepo{
		Did:           did,
		Knot:          knot,
		Repo:          rkeyPtr,
		RepoDid:       &repoDid,
		DefaultBranch: defaultBranch,
	}
}

func (s *Spindle) resolvePipelineSourceRepo(ctx context.Context, trigger *workflow.TriggerMetadata) (*workflow.TriggerRepo, error) {
	if trigger == nil {
		return nil, nil
	}
	if trigger.SourceRepo == nil || *trigger.SourceRepo == "" {
		return trigger.Repo, nil
	}
	repoDid, err := syntax.ParseDID(*trigger.SourceRepo)
	if err != nil {
		return nil, fmt.Errorf("parse sourceRepo %s: %w", *trigger.SourceRepo, err)
	}
	return s.resolveSourceRepoInfo(ctx, repoDid)
}

func (s *Spindle) resolveSourceRepoInfo(ctx context.Context, repoDid syntax.DID) (*workflow.TriggerRepo, error) {
	repo, err := s.db.GetRepoByDid(repoDid)
	if err == nil {
		return s.buildTriggerRepo(ctx, repo)
	}

	// the knot comes from the did doc, not the caller, or the source could point anywhere
	k, err := s.knots.KnotOf(ctx, repoDid)
	if err != nil {
		return nil, fmt.Errorf("verify sourceRepo %s: %w", repoDid, err)
	}
	var manifest org_tangled.RepoManifest
	if err := s.knots.GetRecord(ctx, k, repoDid, org_tangled.RepoManifestNSID, knotclient.ManifestRkey, &manifest); err != nil {
		return nil, fmt.Errorf("verify sourceRepo %s: reading manifest: %w", repoDid, err)
	}
	owner, _, rkey, err := manifestIdentity(&manifest)
	if err != nil {
		return nil, fmt.Errorf("verify sourceRepo %s: %w", repoDid, err)
	}
	declared, err := s.pdsDeclaredRepo(ctx, owner, rkey.String())
	if err != nil {
		return nil, fmt.Errorf("verify sourceRepo %s: reading declaration %s/%s: %w", repoDid, owner, rkey, err)
	}
	if declared != repoDid {
		return nil, fmt.Errorf("verify sourceRepo %s: %s/%s declares %s", repoDid, owner, rkey, declared)
	}
	return s.buildTriggerRepoFrom(ctx, k.Host, owner.String(), rkey.String(), repoDid.String()), nil
}

func (s *Spindle) runPipeline(ctx context.Context, pipelineId models.PipelineId, repoDid syntax.DID, trigger workflow.TriggerMetadata, changedFiles []string, repoCloneUri, repoPath, rev string, only []string, sourceRepo *workflow.TriggerRepo, blocked bool) (models.PipelineId, error) {
	l := log.FromContext(ctx)

	// all triggers go through this gate, not just pushes
	if gate, err := s.db.Gate(&s.cfg.Server, repoDid); err != nil {
		return "", fmt.Errorf("checking admission: %w", err)
	} else if gate.Paused {
		l.Info("owner isn't admitted, starting nothing for their repo", "repo", repoDid)
		return "", nil
	}

	compiler := workflow.Compiler{
		ChangedFiles: changedFiles,
		Trigger:      trigger,
	}

	rawPipeline, _, err := s.loadPipeline(ctx, repoCloneUri, repoPath, rev)
	if err != nil {
		return "", fmt.Errorf("loading pipeline: %w", err)
	}
	if len(rawPipeline) == 0 {
		return "", nil
	}

	tpl := compiler.Compile(compiler.Parse(rawPipeline))
	// todo(dawn): pass compile error to workflow log
	for _, w := range compiler.Diagnostics.Errors {
		l.Error(w.String())
	}
	for _, w := range compiler.Diagnostics.Warnings {
		l.Warn(w.String())
	}

	if len(only) > 0 {
		tpl.Workflows = filterWorkflows(tpl.Workflows, only)
	}
	if len(tpl.Workflows) == 0 {
		return "", nil
	}

	record, err := pipelinecodec.FromTangled(pipelineId, time.Now(), tpl)
	if err != nil {
		return "", fmt.Errorf("creating pipeline: %w", err)
	}
	err = s.processPipeline(ctx, repoDid, tpl, record, sourceRepo, blocked)
	// both errors mean nothing to enqueue, since a pause or wipe can land mid checkout
	if errors.Is(err, db.ErrOwnerPaused) || errors.Is(err, db.ErrNotServed) {
		l.Info("nothing enqueued, the owner is paused or the repo is gone", "repo", repoDid)
		return "", nil
	}
	if err != nil {
		return "", err
	}
	return pipelineId, nil
}

func filterWorkflows(workflows []*workflow.WorkflowSpec, only []string) []*workflow.WorkflowSpec {
	allowed := make(map[string]struct{}, len(only))
	for _, n := range only {
		allowed[n] = struct{}{}
	}
	var filtered []*workflow.WorkflowSpec
	for _, w := range workflows {
		if w == nil {
			continue
		}
		if _, ok := allowed[w.Name]; ok {
			filtered = append(filtered, w)
		}
	}
	return filtered
}

// TriggerManual dispatches a pipeline at sha, authorized against and recorded
// under repoDid. sourceRepo, pull, and inputs are optional trigger payload.
func (s *Spindle) TriggerManual(ctx context.Context, actor, repoDid syntax.DID, sha, ref string, workflows []string, inputs []*workflow.TriggerPair) (models.PipelineId, error) {
	repo, err := s.db.GetRepoByDid(repoDid)
	if err != nil {
		return "", fmt.Errorf("unknown repoDid %s: %w", repoDid, err)
	}

	if ban, err := s.db.IsBanned(repoDid, repo.Owner); err != nil {
		return "", fmt.Errorf("checking bans: %w", err)
	} else if ban != nil {
		return "", fmt.Errorf("repo %s is banned", repoDid)
	}
	// checked before the checkout, otherwise a paused owner matches nothing instead of erroring
	if gate, err := s.db.Gate(&s.cfg.Server, repoDid); err != nil {
		return "", fmt.Errorf("checking admission: %w", err)
	} else if gate.Paused {
		return "", xrpc.ErrOwnerPaused
	}

	triggerRepo, err := s.buildTriggerRepo(ctx, repo)
	if err != nil {
		return "", fmt.Errorf("building trigger repo: %w", err)
	}

	var refPtr *string
	if ref != "" {
		refPtr = &ref
	}
	trigger := workflow.TriggerMetadata{
		Kind:  string(workflow.TriggerKindManual),
		Repo:  triggerRepo,
		Actor: (*string)(&actor),
		Manual: &workflow.ManualTriggerData{
			Sha:    sha,
			Ref:    refPtr,
			Inputs: inputs,
		},
	}

	repoCloneUri, repoPath := s.resolveCheckout(repo)

	// a manual ref only becomes a signable claim once the server has seen the
	// sha inside the ref's history
	if refPtr != nil {
		contains, err := gitutil.RefContainsCommit(ctx, repoCloneUri, ref, sha)
		if err != nil {
			return "", fmt.Errorf("checking ref %s: %w", ref, err)
		}
		if !contains {
			return "", fmt.Errorf("ref %s does not contain commit %s", ref, sha)
		}
	}

	pipelineId, err := s.runPipeline(ctx, models.PipelineId(tid.TID()), repoDid, trigger, nil, repoCloneUri, repoPath, sha, workflows, triggerRepo, false)
	if err != nil {
		return "", err
	}
	if pipelineId == "" {
		return "", xrpc.ErrNoMatchingWorkflows
	}
	return pipelineId, nil
}

func (s *Spindle) resolveCheckout(repo *db.Repo) (cloneUri, repoPath string) {
	return s.knots.CloneURL(knotclient.Knot{Host: repo.Knot}, repo.RepoDid), s.newRepoPath(repo.RepoDid)
}

// resolves the workflow definition at sha without executing it
// returns a deterministic fingerprint over the resolved files.
func (s *Spindle) ListWorkflowDefinitions(ctx context.Context, repoDid syntax.DID, ref string) ([]*models.WorkflowDefinition, error) {
	if ref == "" {
		ref = "HEAD"
	}
	return s.workflowDefs.list(repoDid, ref, func() ([]*models.WorkflowDefinition, error) {
		return s.loadWorkflowDefinitions(ctx, repoDid, ref)
	})
}

func (s *Spindle) loadWorkflowDefinitions(ctx context.Context, repoDid syntax.DID, ref string) ([]*models.WorkflowDefinition, error) {
	rawPipeline, commit, err := s.readWorkflowDir(ctx, repoDid, ref)
	if err != nil {
		return nil, fmt.Errorf("loading pipeline: %w", err)
	}
	out := make([]*models.WorkflowDefinition, 0, len(rawPipeline))
	for _, raw := range rawPipeline {
		definition, err := pipelinecodec.FileDefinition(raw.Name, raw.Contents, repoDid.String(), commit)
		if err != nil {
			return nil, fmt.Errorf("parsing workflow %s: %w", raw.Name, err)
		}
		out = append(out, definition)
	}
	return out, nil
}

// the host comes from a did doc the repo owner controls and git dials it unguarded, so
// every knot fetch comes through here and git's stderr only goes to the log
func (s *Spindle) syncRepo(ctx context.Context, cloneUri, repoPath, rev string) error {
	err := s.guardedSync(ctx, cloneUri, repoPath, rev)
	if err != nil {
		log.FromContext(ctx).Error("sync git repo", "uri", cloneUri, "err", err)
		if refusedByKnot(err) {
			return errRevRefused
		}
		return errors.New("couldn't fetch the push from the knot")
	}
	return nil
}

var (
	errRevRefused = errors.New("the knot refused to serve that commit")
	// git only passes on the status of a refused smart http request, not the knot's reason
	gitHTTPStatus = regexp.MustCompile(`returned error: (\d{3})`)
)

// a 4xx means the knot won't serve that commit, a retry gets the same answer unless it was busy
func refusedByKnot(err error) bool {
	msg := err.Error()
	// upload-pack answers 200 for an unknown sha, which leaves only the message to match
	if strings.Contains(msg, "not our ref") || strings.Contains(msg, "unadvertised object") {
		return true
	}
	// info/refs 404s only show up in the message text
	if strings.Contains(msg, "fatal: repository '") && strings.Contains(msg, "' not found") {
		return true
	}
	m := gitHTTPStatus.FindStringSubmatch(msg)
	if m == nil {
		return false
	}
	status, _ := strconv.Atoi(m[1])
	return status >= 400 && status < 500 && status != http.StatusRequestTimeout && status != http.StatusTooManyRequests
}

func (s *Spindle) guardedSync(ctx context.Context, cloneUri, repoPath, rev string) error {
	// a plain path has no host to dial
	if u, err := url.Parse(cloneUri); err == nil && u.Host != "" && !s.cfg.Server.Dev {
		if err := netguard.CheckHost(ctx, u.Host); err != nil {
			return err
		}
	}
	return gitutil.SparseSync(ctx, cloneUri, repoPath, rev, sparseWorkflowDir)
}

func (s *Spindle) loadPipeline(ctx context.Context, repoUri, repoPath, rev string) (workflow.RawPipeline, string, error) {
	if err := s.syncRepo(ctx, repoUri, repoPath, rev); err != nil {
		return nil, "", err
	}
	// go-git chokes on sha256 repos, so git resolves the commit
	head, err := gitutil.Output(ctx, "-C", repoPath, "rev-parse", "HEAD")
	if err != nil {
		return nil, "", fmt.Errorf("resolving checked out commit: %w", err)
	}
	commit := strings.TrimSpace(string(head))
	rawPipeline, err := readWorkflowTree(ctx, gitutil.Output, repoPath, commit)
	return rawPipeline, commit, err
}

func (s *Spindle) newRepoPath(repo syntax.DID) string {
	return filepath.Join(s.cfg.Server.RepoDir, repo.String())
}

const RequiredVersion = "2.49.0"

func ensureGitVersion() error {
	v, err := gitutil.Version()
	if err != nil {
		return fmt.Errorf("fetching git version: %w", err)
	}
	if v.LessThan(version.Must(version.NewVersion(RequiredVersion))) {
		return fmt.Errorf("installed git version %q is not supported, Spindle requires git version >= %q", v, RequiredVersion)
	}
	return nil
}

func (s *Spindle) configureOwner() error {
	cfgOwner := s.cfg.Server.Owner

	existing, err := s.e.GetSpindleUsersByRole("server:owner", rbacDomain)
	if err != nil {
		return err
	}

	switch len(existing) {
	case 0:
		// no owner configured, continue
	case 1:
		// find existing owner
		existingOwner := existing[0]

		// no ownership change, this is okay
		if existingOwner == s.cfg.Server.Owner {
			break
		}

		// remove existing owner
		err = s.e.RemoveSpindleOwner(rbacDomain, existingOwner)
		if err != nil {
			return nil
		}
	default:
		return fmt.Errorf("more than one owner in DB, try deleting %q and starting over", s.cfg.Server.DBPath)
	}

	return s.e.AddSpindleOwner(rbacDomain, cfgOwner)
}

func (s *Spindle) StartJobWorkers(ctx context.Context) {
	for range s.cfg.Server.MaxJobCount {
		s.jobWorkers.Add(1)
		go func() {
			defer s.jobWorkers.Done()
			for {
				job, err := s.dequeueJob(ctx)
				if err != nil && ctx.Err() == nil {
					s.l.Error("failed to dequeue job", "error", err)
				}
				if job == nil {
					// polls because a failed dequeue can leave jobs queued with nobody to wake us
					var retry <-chan time.Time
					if err != nil {
						retry = time.After(dequeueRetryDelay)
					}
					select {
					case <-ctx.Done():
						return
					case <-s.jobWake:
					case <-retry:
					}
					continue
				}
				s.metrics.RecordJobQueueActivity("dequeue")
				s.runJob(ctx, job)
				s.finishJob(job.Owner)
			}
		}()
	}
}

func (s *Spindle) dequeueJob(ctx context.Context) (*db.JobRow, error) {
	s.jobsMu.Lock()
	defer s.jobsMu.Unlock()
	job, err := s.db.DequeueJob(ctx, s.jobsByOwner)
	if job != nil {
		if s.jobsByOwner == nil {
			s.jobsByOwner = make(map[string]int)
		}
		s.jobsByOwner[job.Owner]++
	}
	return job, err
}

func (s *Spindle) finishJob(owner string) {
	s.jobsMu.Lock()
	defer s.jobsMu.Unlock()
	s.jobsByOwner[owner]--
	if s.jobsByOwner[owner] <= 0 {
		delete(s.jobsByOwner, owner)
	}
}

func (s *Spindle) runJob(ctx context.Context, job *db.JobRow) {
	pipelineId := job.PipelineId
	repoDID := job.RepoDid
	if job.SourceRepo != nil && job.SourceRepo.RepoDid != nil && *job.SourceRepo.RepoDid != "" {
		repoDID = *job.SourceRepo.RepoDid
	}

	jobCtx := observability.ExtractFromTraceparentAndTracestate(ctx, job.Traceparent, job.Tracestate)
	if job.CreatedAtNs > 0 {
		queueDelay := time.Since(time.Unix(0, job.CreatedAtNs))
		if queueDelay < 0 {
			queueDelay = 0
		}
		s.metrics.RecordJobDequeueLatency(jobCtx, queueDelay)
	}

	jobCtx, span := observability.Tracer().Start(jobCtx, "job.run")
	if span.IsRecording() {
		attrs := []attribute.KeyValue{
			attribute.Int64(observability.JobIDKey, job.Id),
			attribute.String(observability.PipelineIDKey, pipelineId.String()),
		}
		if repoDID != "" {
			attrs = append(attrs, attribute.String(observability.RepoDIDKey, repoDID))
		}
		if job.RepoDid != "" && job.RepoDid != repoDID {
			attrs = append(attrs, attribute.String(observability.TargetRepoDIDKey, job.RepoDid))
		}
		span.SetAttributes(attrs...)
	}
	failed := false
	defer func() {
		if failed {
			span.SetStatus(codes.Error, "job failed")
		} else {
			span.SetStatus(codes.Ok, "success")
		}
		span.End()
	}()

	l := log.SubLogger(log.SubLogger(s.l, "job"), "engine").With(
		"job_id", job.Id,
	)
	if job.RepoDid != "" && job.RepoDid != repoDID {
		l = l.With("target_repo_did", job.RepoDid)
	}

	pipelineEnv := models.PipelineEnvVarsForSource(job.Tpl.TriggerMetadata, pipelineId, job.SourceRepo)

	initTpl := job.Tpl
	if job.SourceRepo != nil && job.Tpl.TriggerMetadata != nil {
		tm := *job.Tpl.TriggerMetadata
		tm.Repo = job.SourceRepo
		initTpl.TriggerMetadata = &tm
	}
	trustedSource := models.TrustedPipelineSource(initTpl.TriggerMetadata, job.RepoDid)

	workflows := make(map[models.Engine][]models.Workflow)
	for _, w := range job.Tpl.Workflows {
		if w == nil {
			continue
		}
		eng, ok := s.engs[w.Engine]
		if !ok {
			failed = true
			_ = s.db.StatusFailed(models.WorkflowId{
				PipelineId: pipelineId,
				Name:       w.Name,
			}, fmt.Sprintf("unknown engine %#v", w.Engine), -1, s.n)
			s.metrics.RecordWorkflowTerminal(
				fmt.Sprint(w.Engine),
				"failure",
				string(engine.FailureClassUser),
				string(engine.FailureReasonWorkflowInvalid),
			)
			continue
		}

		ewf, err := eng.InitWorkflow(*w, initTpl)
		if err != nil {
			failed = true
			_ = s.db.StatusFailed(models.WorkflowId{
				PipelineId: pipelineId,
				Name:       w.Name,
			}, fmt.Sprintf("init workflow: %s", err), -1, s.n)
			failureClass, failureReason := engine.FailureAttribution("failure", err)
			s.metrics.RecordWorkflowTerminal(
				fmt.Sprint(w.Engine),
				"failure",
				failureClass,
				failureReason,
			)
			continue
		}
		ewf.RunID = fmt.Sprintf("%d", job.Id)

		if ewf.Environment == nil {
			ewf.Environment = make(map[string]string)
		}
		maps.Copy(ewf.Environment, pipelineEnv)

		ewf.Engine = w.Engine
		ewf.RepoDID = job.RepoDid
		if w.Audience != nil {
			ewf.Audience = *w.Audience
		}
		if !trustedSource {
			ewf.Caches = nil
		}
		workflows[eng] = append(workflows[eng], *ewf)
	}

	engine.StartWorkflows(l, s.vault, s.cfg, s.qm, s.stores, s.db, s.n, s.cache, s.cacheCtl, jobCtx, &models.Pipeline{
		RepoDid:         syntax.DID(job.RepoDid),
		Workflows:       workflows,
		TrustedSource:   trustedSource,
		TriggerMetadata: job.Tpl.TriggerMetadata,
		MintIDToken:     s.idTokenMinter(job, trustedSource),
		GrantNixCache:   s.nixCacheGranter(job, trustedSource),
	}, pipelineId)
}

// only trusted source runs push to the repo's cache, and attic records the run as the uploader
func (s *Spindle) nixCacheGranter(job *db.JobRow, trustedSource bool) func(context.Context, models.WorkflowId, time.Duration) (*models.NixCacheGrant, error) {
	if s.nixCache == nil {
		return nil
	}
	return func(ctx context.Context, wid models.WorkflowId, workflowTimeBudget time.Duration) (*models.NixCacheGrant, error) {
		sub := string(wid.PipelineId) + "/" + wid.Name
		return s.nixCache.Grant(ctx, job.RepoDid, job.Owner, sub, trustedSource, time.Now().Add(workflowTimeBudget+nixCacheTokenGrace))
	}
}

// only trusted source runs receive a token. unrecordable identities withhold
// it; signing failures fail the workflow.
func (s *Spindle) idTokenMinter(job *db.JobRow, trustedSource bool) func(time.Duration, string) (string, error) {
	if s.oidc == nil || !trustedSource || job.Tpl.TriggerMetadata == nil {
		return nil
	}

	identity := models.RunIdentityFor(job.Tpl.TriggerMetadata, job.RepoDid)
	if identity.SHA == "" {
		s.l.Warn("id tokens are off for a run with no recorded commit", "job_id", job.Id)
		return nil
	}

	run := oidc.Run{
		Repository: identity.Repository,
		Ref:        identity.Ref,
		SHA:        identity.SHA,
		Actor:      identity.Actor,
	}
	return func(workflowTimeBudget time.Duration, audience string) (string, error) {
		token, err := s.oidc.Mint(run, workflowTimeBudget, audience)
		if errors.Is(err, oidc.ErrUnrecordableRun) {
			s.l.Warn("withholding an id token the issuer will not sign", "job_id", job.Id, "err", err)
			return "", nil
		}
		return token, err
	}
}

// reruns the spec a pipeline was enqueued with under a new id, on behalf of actor. a pipeline
// that was blocked has to be approved first, even once cancelled. a retry would skip the approval
func (s *Spindle) RetryPipeline(ctx context.Context, actor, repoDid syntax.DID, id models.PipelineId, workflows []string) (models.PipelineId, error) {
	stored, err := s.db.GetStoredPipeline(id)
	if err != nil {
		return "", err
	}
	if stored.Spec == nil || stored.Spec.TriggerMetadata == nil || stored.NeededApproval && !stored.Approved {
		return "", xrpc.ErrNotRetryable
	}
	tpl := *stored.Spec
	trigger := *tpl.TriggerMetadata
	trigger.Actor = (*string)(&actor)
	if stored.NeededApproval && trigger.PullRequest != nil {
		pull := *trigger.PullRequest
		pull.SourceBranch = ""
		trigger.PullRequest = &pull
	}
	tpl.TriggerMetadata = &trigger
	if len(workflows) > 0 {
		tpl.Workflows = filterWorkflows(tpl.Workflows, workflows)
	}
	if len(tpl.Workflows) == 0 {
		return "", xrpc.ErrNoMatchingWorkflows
	}

	retryId := models.PipelineId(tid.TID())
	record, err := pipelinecodec.FromTangled(retryId, time.Now(), tpl)
	if err != nil {
		return "", fmt.Errorf("creating pipeline: %w", err)
	}
	if err := s.processPipeline(ctx, repoDid, tpl, record, stored.SourceRepo, false); err != nil {
		return "", err
	}
	return retryId, nil
}

// the guards run again, an owner paused or banned since the pipeline was blocked stays blocked
func (s *Spindle) ApprovePipeline(ctx context.Context, repoDid syntax.DID, id models.PipelineId) error {
	traceparent, tracestate := observability.InjectToTraceparentAndTracestate(ctx)
	err := func() error {
		defer s.lockRepo(repoDid).Unlock()
		return s.db.ApprovePipeline(s.rootCtx, repoDid.String(), id, traceparent, tracestate, s.n, db.Admitted(&s.cfg.Server))
	}()
	if err != nil {
		return err
	}
	s.l.Info("pipeline approved", "id", id)

	select {
	case s.jobWake <- struct{}{}:
	default:
	}
	return nil
}

func (s *Spindle) processPipeline(ctx context.Context, repoDid syntax.DID, tpl workflow.PipelineSpec, record *models.PipelineRecord, sourceRepo *workflow.TriggerRepo, blocked bool) error {
	traceparent, tracestate := observability.InjectToTraceparentAndTracestate(ctx)
	// only the insert holds the repo lock, which keeps a wipe from landing halfway through
	err := func() error {
		defer s.lockRepo(repoDid).Unlock()
		return s.db.EnqueuePipeline(s.rootCtx, repoDid.String(), record, sourceRepo, tpl, blocked, traceparent, tracestate, s.n, db.Admitted(&s.cfg.Server))
	}()
	if err != nil {
		return fmt.Errorf("failed to enqueue durable job: %w", err)
	}
	if blocked {
		s.l.Info("pipeline is blocked until approved", "id", record.ID)
		return nil
	}
	s.l.Info("pipeline enqueued successfully to db", "id", record.ID)

	select {
	case s.jobWake <- struct{}{}:
	default:
	}

	return nil
}