package spindle

import (
	"container/heap"
	"context"
	"fmt"
	"log/slog"
	"slices"
	"sync"
	"time"

	"github.com/bluesky-social/indigo/atproto/syntax"
	"github.com/go-git/go-git/v5/plumbing"
	"github.com/robfig/cron/v3"
	"tangled.org/core/api/org_tangled"
	"tangled.org/core/gitclient"
	"tangled.org/core/log"
	"tangled.org/core/spindle/db"
	"tangled.org/core/spindle/knotclient"
	"tangled.org/core/spindle/models"
	"tangled.org/core/tid"
	"tangled.org/core/workflow"
)

const (
	scheduleRunPruneInterval = 6 * time.Hour
	scheduleDispatchBacklog  = 60
	// sized so only a genuinely stalled fetch can hit it; a hung dispatch
	// would otherwise wedge the single-minute dispatch loop
	scheduleDispatchTimeout = 10 * time.Minute
)

type scheduledWorkflow struct {
	repo          db.Repo
	scheduledRepo db.ScheduledRepo
	name          string
	expression    string
	timezone      string
	schedule      cron.Schedule
	next          time.Time
	index         int
}

type scheduleHeap []*scheduledWorkflow

func (h scheduleHeap) Len() int           { return len(h) }
func (h scheduleHeap) Less(i, j int) bool { return h[i].next.Before(h[j].next) }
func (h scheduleHeap) Swap(i, j int) {
	h[i], h[j] = h[j], h[i]
	h[i].index = i
	h[j].index = j
}
func (h *scheduleHeap) Push(value any) {
	entry := value.(*scheduledWorkflow)
	entry.index = len(*h)
	*h = append(*h, entry)
}
func (h *scheduleHeap) Pop() any {
	old := *h
	last := len(old) - 1
	entry := old[last]
	old[last] = nil
	entry.index = -1
	*h = old[:last]
	return entry
}

type scheduleLoader func(context.Context, db.Repo, time.Time) ([]scheduledWorkflow, db.ScheduledRepo, error)
type scheduleDispatcher func(context.Context, db.Repo, db.ScheduledRepo, []string, time.Time) (models.PipelineId, error)

type pipelineScheduler struct {
	db       *db.DB
	l        *slog.Logger
	load     scheduleLoader
	dispatch scheduleDispatcher
	now      func() time.Time

	minimumInterval time.Duration
	concurrency     int
	dispatchTimeout time.Duration

	mu     sync.Mutex
	byRepo map[string][]*scheduledWorkflow
	queue  scheduleHeap

	// don't hold refreshMu across a clone
	refreshMu  sync.Mutex
	inflight   map[string]*refreshEntry
	dispatchMu sync.Mutex
}

// one waiter at most, a newer one replaces it
type refreshEntry struct {
	slot     chan struct{}
	users    int
	removals uint64
	waiter   *refreshWaiter
}

type refreshWaiter struct {
	// removals seen when it asked, same count later means nothing was removed since
	at         uint64
	superseded chan struct{}
}

func newPipelineScheduler(s *Spindle) *pipelineScheduler {
	minimumInterval := s.cfg.Schedule.MinimumInterval
	concurrency := s.cfg.Schedule.Concurrency
	if concurrency < 1 {
		concurrency = 1
	}
	return &pipelineScheduler{
		db:              s.db,
		l:               log.SubLogger(s.l, "schedule"),
		load:            s.loadRepoSchedules,
		dispatch:        s.dispatchScheduledPipeline,
		now:             time.Now,
		minimumInterval: minimumInterval,
		concurrency:     concurrency,
		dispatchTimeout: scheduleDispatchTimeout,
		byRepo:          make(map[string][]*scheduledWorkflow),
	}
}

func (s *pipelineScheduler) Start(ctx context.Context) {
	if err := s.restoreSchedules(ctx); err != nil {
		s.l.Error("failed to restore schedules", "err", err)
	}
	go s.backfillSchedules(ctx)
	go s.run(ctx)
}

func (s *pipelineScheduler) run(ctx context.Context) {
	dispatches := make(chan time.Time, scheduleDispatchBacklog)
	go s.runDispatches(ctx, dispatches)

	if err := s.db.PruneScheduleRuns(ctx, s.schedulePruneBefore(s.now())); err != nil {
		s.l.Error("failed to prune schedule claims", "err", err)
	}
	pruneTicker := time.NewTicker(scheduleRunPruneInterval)
	defer pruneTicker.Stop()

	timer := time.NewTimer(time.Until(nextScheduleMinute(s.now())))
	defer timer.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case firedAt := <-timer.C:
			scheduledAt := firedAt.UTC().Truncate(time.Minute)
			select {
			case dispatches <- scheduledAt:
			default:
				s.l.Error("schedule dispatch backlog is full", "scheduledAt", scheduledAt)
			}
			timer.Reset(time.Until(nextScheduleMinute(s.now())))
		case <-pruneTicker.C:
			if err := s.db.PruneScheduleRuns(ctx, s.schedulePruneBefore(s.now())); err != nil {
				s.l.Error("failed to prune schedule claims", "err", err)
			}
		}
	}
}

func (s *pipelineScheduler) effectiveDispatchTimeout() time.Duration {
	if s.dispatchTimeout > 0 {
		return s.dispatchTimeout
	}
	return scheduleDispatchTimeout
}

func (s *pipelineScheduler) schedulePruneBefore(now time.Time) time.Time {
	retention := 24 * time.Hour
	if s.minimumInterval > retention {
		retention = s.minimumInterval
	}
	return now.UTC().Add(-retention)
}

func nextScheduleMinute(now time.Time) time.Time {
	return now.UTC().Truncate(time.Minute).Add(time.Minute)
}

func (s *pipelineScheduler) runDispatches(ctx context.Context, dispatches <-chan time.Time) {
	for {
		select {
		case <-ctx.Done():
			return
		case scheduledAt, ok := <-dispatches:
			if !ok {
				return
			}
			s.dispatchDue(ctx, scheduledAt)
		}
	}
}

func (s *pipelineScheduler) restoreSchedules(ctx context.Context) error {
	repos, err := s.db.AllRepos()
	if err != nil {
		return fmt.Errorf("listing repositories: %w", err)
	}
	repoByDid := make(map[string]db.Repo, len(repos))
	for _, repo := range repos {
		repoByDid[repo.RepoDid.String()] = repo
	}

	definitions, err := s.db.WorkflowSchedules(ctx)
	if err != nil {
		return fmt.Errorf("listing workflow schedules: %w", err)
	}
	grouped := make(map[string][]scheduledWorkflow)
	groupedSchedules := make(map[string]map[string][]cron.Schedule)
	seen := make(map[struct {
		repo       string
		workflow   string
		expression string
		timezone   string
	}]struct{})
	for _, definition := range definitions {
		repo, ok := repoByDid[definition.RepoDid]
		if !ok {
			continue
		}
		if _, ok := grouped[definition.RepoDid]; !ok {
			grouped[definition.RepoDid] = nil
		}
		timezone := definition.Timezone
		if timezone == "" {
			timezone = "UTC"
		}
		key := struct {
			repo       string
			workflow   string
			expression string
			timezone   string
		}{definition.RepoDid, definition.Workflow, definition.Expression, timezone}
		if _, ok := seen[key]; ok {
			continue
		}
		seen[key] = struct{}{}
		schedule, err := workflow.ParseCron(
			definition.Expression,
			timezone,
			definition.RepoDid+"\x00"+definition.Workflow,
		)
		if err != nil {
			s.l.Warn("ignoring invalid persisted schedule", "repo", definition.RepoDid, "workflow", definition.Workflow, "expression", definition.Expression, "timezone", timezone, "err", err)
			continue
		}
		grouped[definition.RepoDid] = append(grouped[definition.RepoDid], scheduledWorkflow{
			repo: repo,
			scheduledRepo: db.ScheduledRepo{
				RepoDid: definition.RepoDid,
				Branch:  definition.Branch,
				SHA:     definition.SHA,
			},
			name:       definition.Workflow,
			expression: definition.Expression,
			timezone:   timezone,
			schedule:   schedule,
		})
		if groupedSchedules[definition.RepoDid] == nil {
			groupedSchedules[definition.RepoDid] = make(map[string][]cron.Schedule)
		}
		groupedSchedules[definition.RepoDid][definition.Workflow] = append(
			groupedSchedules[definition.RepoDid][definition.Workflow],
			schedule,
		)
	}
	now := s.now()
	for repoDid, entries := range grouped {
		invalidWorkflows := make(map[string]struct{})
		for workflowName, schedules := range groupedSchedules[repoDid] {
			if err := workflow.ValidateScheduleInterval(schedules, s.minimumInterval, now); err != nil {
				s.l.Warn("ignoring invalid persisted workflow schedules", "repo", repoDid, "workflow", workflowName, "minimumInterval", s.minimumInterval, "err", err)
				invalidWorkflows[workflowName] = struct{}{}
			}
		}
		valid := make([]scheduledWorkflow, 0, len(entries))
		for _, entry := range entries {
			if _, invalid := invalidWorkflows[entry.name]; invalid {
				continue
			}
			valid = append(valid, entry)
		}
		s.replaceCachedSchedules(repoDid, valid, now)
	}
	s.l.Info("restored persisted schedules", "repositories", len(grouped), "schedules", len(definitions))
	return nil
}

func (s *pipelineScheduler) backfillSchedules(ctx context.Context) {
	repos, err := s.db.UnindexedScheduleRepos(ctx)
	if err != nil {
		s.l.Error("failed to list repositories needing schedule backfill", "err", err)
		return
	}
	if len(repos) == 0 {
		return
	}
	workers := s.workerLimit(len(repos))
	jobs := make(chan db.Repo)
	var wg sync.WaitGroup
	wg.Add(workers)
	for range workers {
		go func() {
			defer wg.Done()
			for {
				select {
				case <-ctx.Done():
					return
				case repo, ok := <-jobs:
					if !ok || ctx.Err() != nil {
						return
					}
					if err := s.RefreshRepo(ctx, repo); err != nil {
						s.l.Warn("failed to backfill repository schedules", "repo", repo.RepoDid, "err", err)
					}
				}
			}
		}()
	}
	for _, repo := range repos {
		select {
		case <-ctx.Done():
			close(jobs)
			wg.Wait()
			return
		case jobs <- repo:
		}
	}
	close(jobs)
	wg.Wait()
}

func (s *pipelineScheduler) workerLimit(work int) int {
	limit := s.concurrency
	if limit < 1 {
		limit = 1
	}
	if work > 0 && limit > work {
		return work
	}
	return limit
}

// read here, and a removal at any point after drops the result
// a covered request is served by the waiter that's already queued
func (s *pipelineScheduler) beginRefresh(repoDid string) (e *refreshEntry, removals uint64, covered bool) {
	s.refreshMu.Lock()
	if s.inflight == nil {
		s.inflight = make(map[string]*refreshEntry)
	}
	e = s.inflight[repoDid]
	if e == nil {
		e = &refreshEntry{slot: make(chan struct{}, 1)}
		s.inflight[repoDid] = e
	}
	if e.waiter != nil && e.waiter.at == e.removals {
		s.refreshMu.Unlock()
		return e, 0, true
	}
	e.users++
	removals = e.removals
	select {
	case e.slot <- struct{}{}:
		s.refreshMu.Unlock()
		return e, removals, false
	default:
	}
	// only one waiter at a time, the newer one has the fresher removal count
	w := &refreshWaiter{at: removals, superseded: make(chan struct{})}
	if e.waiter != nil {
		close(e.waiter.superseded)
	}
	e.waiter = w
	s.refreshMu.Unlock()

	select {
	case e.slot <- struct{}{}:
		s.refreshMu.Lock()
		if e.waiter == w {
			e.waiter = nil
		}
		s.refreshMu.Unlock()
		return e, removals, false
	case <-w.superseded:
		s.leaveRefresh(repoDid, e)
		return e, 0, true
	}
}

func (s *pipelineScheduler) leaveRefresh(repoDid string, e *refreshEntry) {
	s.refreshMu.Lock()
	defer s.refreshMu.Unlock()
	if e.users--; e.users == 0 {
		delete(s.inflight, repoDid)
	}
}

func (s *pipelineScheduler) endRefresh(repoDid string, e *refreshEntry) {
	<-e.slot
	s.leaveRefresh(repoDid, e)
}

func (s *pipelineScheduler) RefreshRepo(ctx context.Context, repo db.Repo) error {
	e, removals, covered := s.beginRefresh(repo.RepoDid.String())
	if covered {
		return nil
	}
	defer s.endRefresh(repo.RepoDid.String(), e)

	current, err := s.db.GetRepoByDid(repo.RepoDid)
	if err != nil {
		s.removeCachedSchedules(repo.RepoDid.String())
		return fmt.Errorf("checking repository registration: %w", err)
	}
	repo = *current
	entries, scheduledRepo, err := s.load(ctx, repo, s.now())
	if err != nil {
		s.removeCachedSchedules(repo.RepoDid.String())
		return err
	}
	if scheduledRepo.RepoDid == "" {
		scheduledRepo.RepoDid = repo.RepoDid.String()
	}
	if scheduledRepo.RepoDid != repo.RepoDid.String() {
		s.removeCachedSchedules(repo.RepoDid.String())
		return fmt.Errorf("loaded schedule snapshot for %s, want %s", scheduledRepo.RepoDid, repo.RepoDid)
	}
	if scheduledRepo.Branch == "" || scheduledRepo.SHA == "" {
		s.removeCachedSchedules(repo.RepoDid.String())
		return fmt.Errorf("loaded schedule snapshot for %s is incomplete", repo.RepoDid)
	}
	definitions := make([]db.WorkflowSchedule, 0, len(entries))
	for i := range entries {
		entries[i].repo = repo
		entries[i].scheduledRepo = scheduledRepo
		definitions = append(definitions, db.WorkflowSchedule{
			RepoDid:    repo.RepoDid.String(),
			Workflow:   entries[i].name,
			Expression: entries[i].expression,
			Timezone:   entries[i].timezone,
			Branch:     scheduledRepo.Branch,
			SHA:        scheduledRepo.SHA,
		})
	}
	refreshedAt := s.now()
	s.refreshMu.Lock()
	defer s.refreshMu.Unlock()
	if e.removals != removals {
		return nil
	}
	if err := s.db.ReplaceWorkflowSchedules(ctx, scheduledRepo, definitions, refreshedAt); err != nil {
		return fmt.Errorf("persisting repository schedules: %w", err)
	}
	s.replaceCachedSchedules(repo.RepoDid.String(), entries, refreshedAt)
	s.l.Debug("repository schedules refreshed", "repo", repo.RepoDid, "count", len(entries))
	return nil
}

// doesn't wait for an inflight clone, those can take minutes
func (s *pipelineScheduler) RemoveRepo(ctx context.Context, repoDid string) error {
	s.refreshMu.Lock()
	defer s.refreshMu.Unlock()
	if e := s.inflight[repoDid]; e != nil {
		e.removals++
	}

	err := s.db.RemoveWorkflowSchedules(ctx, repoDid)
	s.removeCachedSchedules(repoDid)
	return err
}

type dueRepository struct {
	repo          db.Repo
	scheduledRepo db.ScheduledRepo
	workflows     map[string]struct{}
}

func (s *pipelineScheduler) replaceCachedSchedules(repoDid string, entries []scheduledWorkflow, after time.Time) {
	s.mu.Lock()
	defer s.mu.Unlock()

	if s.byRepo == nil {
		s.byRepo = make(map[string][]*scheduledWorkflow)
	}
	s.dropCached(repoDid)

	after = after.UTC().Truncate(time.Minute)
	cached := make([]*scheduledWorkflow, 0, len(entries))
	for i := range entries {
		entry := new(scheduledWorkflow)
		*entry = entries[i]
		entry.next = entry.schedule.Next(after)
		if entry.next.IsZero() || !entry.next.After(after) {
			continue
		}
		heap.Push(&s.queue, entry)
		cached = append(cached, entry)
	}
	if len(cached) > 0 {
		s.byRepo[repoDid] = cached
	}
}

func (s *pipelineScheduler) removeCachedSchedules(repoDid string) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.dropCached(repoDid)
}

// the caller holds mu
func (s *pipelineScheduler) dropCached(repoDid string) {
	for _, entry := range s.byRepo[repoDid] {
		if entry.index >= 0 {
			heap.Remove(&s.queue, entry.index)
		}
	}
	delete(s.byRepo, repoDid)
}

func (s *pipelineScheduler) due(at time.Time) map[string]dueRepository {
	at = at.UTC().Truncate(time.Minute)
	s.mu.Lock()
	defer s.mu.Unlock()

	due := make(map[string]dueRepository)
	for s.queue.Len() > 0 {
		entry := s.queue[0]
		if entry.next.After(at) {
			break
		}
		heap.Pop(&s.queue)
		if entry.next.Equal(at) {
			key := entry.repo.RepoDid.String()
			group := due[key]
			if group.workflows == nil {
				group = dueRepository{
					repo:          entry.repo,
					scheduledRepo: entry.scheduledRepo,
					workflows:     make(map[string]struct{}),
				}
			} else if group.scheduledRepo.Branch == "" && group.scheduledRepo.SHA == "" {
				group.scheduledRepo = entry.scheduledRepo
			}
			group.workflows[entry.name] = struct{}{}
			due[key] = group
		}
		entry.next = entry.schedule.Next(at)
		if entry.next.IsZero() || !entry.next.After(at) {
			continue
		}
		heap.Push(&s.queue, entry)
	}
	return due
}

func (s *pipelineScheduler) dispatchDue(ctx context.Context, at time.Time) {
	at = at.UTC().Truncate(time.Minute)
	s.dispatchMu.Lock()
	defer s.dispatchMu.Unlock()

	due := s.due(at)
	if len(due) == 0 {
		return
	}

	workers := s.workerLimit(len(due))
	jobs := make(chan dueRepository)
	var wg sync.WaitGroup
	wg.Add(workers)
	for range workers {
		go func() {
			defer wg.Done()
			for {
				select {
				case <-ctx.Done():
					return
				case group, ok := <-jobs:
					if !ok || ctx.Err() != nil {
						return
					}
					s.dispatchRepository(ctx, group, at)
				}
			}
		}()
	}
send:
	for _, group := range due {
		select {
		case <-ctx.Done():
			break send
		case jobs <- group:
		}
	}
	close(jobs)
	wg.Wait()
}

func (s *pipelineScheduler) dispatchRepository(ctx context.Context, group dueRepository, at time.Time) {
	repoDid := group.repo.RepoDid.String()
	names := make([]string, 0, len(group.workflows))
	for name := range group.workflows {
		names = append(names, name)
	}
	slices.Sort(names)

	claimed := names[:0]
	for _, name := range names {
		ok, err := s.db.ClaimScheduleRun(ctx, repoDid, name, at, s.minimumInterval)
		if err != nil {
			s.l.Error("failed to claim scheduled workflow", "repo", repoDid, "workflow", name, "scheduledAt", at, "err", err)
			continue
		}
		if ok {
			claimed = append(claimed, name)
		}
	}
	if len(claimed) == 0 {
		return
	}

	dispatchCtx, cancel := context.WithTimeout(ctx, s.effectiveDispatchTimeout())
	defer cancel()
	pipelineID, err := s.dispatch(dispatchCtx, group.repo, group.scheduledRepo, claimed, at)
	if err != nil {
		s.l.Error("scheduled pipeline dispatch failed", "repo", repoDid, "scheduledAt", at, "err", err)
		if pipelineID == "" {
			cleanupCtx := ctx
			if ctx.Err() != nil {
				cleanupCtx = context.WithoutCancel(ctx)
			}
			for _, name := range claimed {
				if releaseErr := s.db.ReleaseScheduleRun(cleanupCtx, repoDid, name, at); releaseErr != nil {
					s.l.Error("failed to release schedule claim", "repo", repoDid, "workflow", name, "scheduledAt", at, "err", releaseErr)
				}
			}
			return
		}
	}

	pipelineAt := ""
	if pipelineID != "" {
		pipelineAt = pipelineID.String()
	}
	cleanupCtx := ctx
	if ctx.Err() != nil {
		cleanupCtx = context.WithoutCancel(ctx)
	}
	for _, name := range claimed {
		if err := s.db.CompleteScheduleRun(cleanupCtx, repoDid, name, at, pipelineAt); err != nil {
			s.l.Error("failed to complete schedule claim", "repo", repoDid, "workflow", name, "scheduledAt", at, "err", err)
		}
	}
}

func (s *Spindle) loadRepoSchedules(ctx context.Context, repo db.Repo, after time.Time) ([]scheduledWorkflow, db.ScheduledRepo, error) {
	branch, tip, err := s.getRepoDefaultBranch(ctx, s.knots.Git(knotclient.Knot{Host: repo.Knot}, repo.RepoDid), repo)
	if err != nil {
		return nil, db.ScheduledRepo{}, err
	}
	scheduledRepo := db.ScheduledRepo{
		RepoDid: repo.RepoDid.String(),
		Branch:  branch,
		SHA:     tip,
	}
	raw, _, err := s.loadPipeline(ctx, s.knots.CloneURL(knotclient.Knot{Host: repo.Knot}, repo.RepoDid), s.newRepoPath(repo.RepoDid), tip)
	if err != nil {
		return nil, scheduledRepo, err
	}

	compiler := workflow.Compiler{}
	parsed := compiler.Parse(raw)
	for _, diagnostic := range compiler.Diagnostics.Errors {
		s.l.Warn("scheduled workflow manifest is invalid", "repo", repo.RepoDid, "diagnostic", diagnostic.String())
	}

	minimumInterval := s.cfg.Schedule.MinimumInterval
	var entries []scheduledWorkflow
	seen := make(map[struct{ name, expression, timezone string }]struct{})
	for _, wf := range parsed {
		var workflowSchedules []cron.Schedule
		var workflowEntries []scheduledWorkflow
		invalid := false
		for _, constraint := range wf.When {
			for _, definition := range constraint.Schedule {
				expression := definition.Cron
				timezone := definition.Timezone
				if timezone == "" {
					timezone = "UTC"
				}
				key := struct{ name, expression, timezone string }{wf.Name, expression, timezone}
				if _, ok := seen[key]; ok {
					continue
				}
				seen[key] = struct{}{}
				schedule, err := workflow.ParseCron(
					expression,
					timezone,
					repo.RepoDid.String()+"\x00"+wf.Name,
				)
				if err != nil {
					s.l.Warn("ignoring invalid workflow schedule", "repo", repo.RepoDid, "workflow", wf.Name, "expression", expression, "timezone", timezone, "err", err)
					invalid = true
					continue
				}
				if !workflow.HasHashedCron(expression) {
					s.l.Warn("scheduled workflow uses exact cron fields; consider H for load spreading", "repo", repo.RepoDid, "workflow", wf.Name, "expression", expression)
				}
				workflowSchedules = append(workflowSchedules, schedule)
				workflowEntries = append(workflowEntries, scheduledWorkflow{
					repo:          repo,
					scheduledRepo: scheduledRepo,
					name:          wf.Name,
					expression:    expression,
					timezone:      timezone,
					schedule:      schedule,
				})
			}
		}
		if invalid {
			s.l.Warn("ignoring workflow with invalid schedule", "repo", repo.RepoDid, "workflow", wf.Name)
			continue
		}
		if err := workflow.ValidateScheduleInterval(workflowSchedules, minimumInterval, after); err != nil {
			s.l.Warn("ignoring workflow with schedules that are too close", "repo", repo.RepoDid, "workflow", wf.Name, "minimumInterval", minimumInterval, "err", err)
			continue
		}
		entries = append(entries, workflowEntries...)
	}
	return entries, scheduledRepo, nil
}

func (s *Spindle) getRepoDefaultBranch(ctx context.Context, git *gitclient.Client, repo db.Repo) (name, hash string, err error) {
	knot := knotclient.Knot{Host: repo.Knot}
	name, err = s.defaultBranch(ctx, knot, repo.RepoDid)
	if err != nil {
		return "", "", err
	}
	oid, err := git.Resolve(ctx, "refs/heads/"+name)
	if err != nil {
		return "", "", fmt.Errorf("resolving branch %s of %s: %w", name, repo.RepoDid, err)
	}
	return name, string(oid), nil
}

// the knot keeps the default branch in its refs record, the manifest doesn't carry it
func (s *Spindle) defaultBranch(ctx context.Context, k knotclient.Knot, repo syntax.DID) (string, error) {
	var refs org_tangled.GitRefs
	if err := s.knots.GetRecord(ctx, k, repo, org_tangled.GitRefsNSID, knotclient.RefsRkey, &refs); err != nil {
		return "", fmt.Errorf("reading refs of %s: %w", repo, err)
	}
	if refs.DefaultBranch == "" {
		return "", fmt.Errorf("%s has no default branch yet", repo)
	}
	return refs.DefaultBranch, nil
}

func (s *Spindle) dispatchScheduledPipeline(ctx context.Context, repo db.Repo, scheduledRepo db.ScheduledRepo, workflows []string, scheduledAt time.Time) (models.PipelineId, error) {
	repoDid := repo.RepoDid.String()
	if scheduledRepo.RepoDid == "" {
		scheduledRepo.RepoDid = repoDid
	}
	if scheduledRepo.RepoDid != repoDid {
		return models.PipelineId(""), fmt.Errorf("scheduled repository snapshot is for %s, want %s", scheduledRepo.RepoDid, repoDid)
	}
	if scheduledRepo.Branch == "" || scheduledRepo.SHA == "" {
		return models.PipelineId(""), fmt.Errorf("scheduled repository snapshot for %s is incomplete", repoDid)
	}
	rkey := repo.Rkey.String()
	ref := plumbing.NewBranchReferenceName(scheduledRepo.Branch).String()
	triggerRepo := &workflow.TriggerRepo{
		Did:           repo.Owner.String(),
		Knot:          repo.Knot,
		Repo:          &rkey,
		RepoDid:       &repoDid,
		DefaultBranch: scheduledRepo.Branch,
	}
	owner := repo.Owner.String()
	trigger := workflow.TriggerMetadata{
		Kind:  string(workflow.TriggerKindSchedule),
		Repo:  triggerRepo,
		Actor: &owner,
		Schedule: &workflow.ScheduleTriggerData{
			Ref:         ref,
			ScheduledAt: scheduledAt.UTC().Truncate(time.Minute).Format(time.RFC3339),
			Sha:         scheduledRepo.SHA,
		},
	}
	return s.runPipeline(
		ctx,
		models.PipelineId(tid.TID()),
		repo.RepoDid,
		trigger,
		nil,
		s.knots.CloneURL(knotclient.Knot{Host: repo.Knot}, repo.RepoDid),
		s.newRepoPath(repo.RepoDid),
		scheduledRepo.SHA,
		workflows,
		triggerRepo,
		false,
	)
}