schedule_tasks.go

// Copyright 2023 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT

package actions

import (
	"context"
	"errors"
	"fmt"
	"maps"
	"time"

	actions_model "gitea.dev/models/actions"
	"gitea.dev/models/db"
	"gitea.dev/models/organization"
	perm_model "gitea.dev/models/perm"
	access_model "gitea.dev/models/perm/access"
	repo_model "gitea.dev/models/repo"
	"gitea.dev/models/unit"
	user_model "gitea.dev/models/user"
	"gitea.dev/modules/json"
	"gitea.dev/modules/log"
	"gitea.dev/modules/timeutil"
	webhook_module "gitea.dev/modules/webhook"
	"gitea.dev/services/convert"
)

// StartScheduleTasks start the task
func StartScheduleTasks(ctx context.Context) error {
	return startTasks(ctx)
}

// startTasks starts every due spec and returns an error if any of them failed.
func startTasks(ctx context.Context) error {
	const pageSize = 50

	now := time.Now()
	var failed int
	var beforeID int64
	for {
		specs, _, err := actions_model.FindSpecs(ctx, actions_model.FindSpecOptions{
			ListOptions: db.ListOptions{Page: 1, PageSize: pageSize},
			Next:        now.Unix(),
			BeforeID:    beforeID,
		})
		if err != nil {
			return fmt.Errorf("find specs: %w", err)
		}

		for _, row := range specs {
			// one failing spec must not abort the pass, or a single broken workflow stops every other schedule
			if err := startTask(ctx, row, now); err != nil {
				failed++
				log.Error("start schedule spec %d (repo %d, schedule %d): %v", row.ID, row.RepoID, row.ScheduleID, err)
			}
		}

		if len(specs) < pageSize {
			break
		}
		beforeID = specs[len(specs)-1].ID // keyset, because advanced specs drop out of the "next <= now" filter
	}

	// surfaces as an admin notice through the cron task, once per occurrence rather than once per pass
	if failed > 0 {
		return fmt.Errorf("%d schedule(s) could not be started", failed)
	}
	return nil
}

// startTask advances the spec to its next occurrence before creating the run, so a failing workflow
// retries on its own schedule instead of on every pass, and a failed update cannot duplicate the run.
func startTask(ctx context.Context, row *actions_model.ActionScheduleSpec, now time.Time) error {
	schedule, err := row.Parse()
	if err != nil {
		return fmt.Errorf("parse %q: %w", row.Spec, err)
	}
	row.Prev = row.Next
	row.Next = timeutil.TimeStamp(schedule.Next(now.Add(time.Minute)).Unix())
	if err := actions_model.UpdateScheduleSpec(ctx, row, "prev", "next"); err != nil {
		return fmt.Errorf("update spec: %w", err)
	}

	if row.Repo == nil || row.Schedule == nil {
		return errors.New("repo or schedule row is gone")
	}
	if row.Repo.IsArchived {
		return nil
	}

	cfg, err := row.Repo.GetUnit(ctx, unit.TypeActions)
	if err != nil {
		if repo_model.IsErrUnitTypeNotExist(err) {
			return nil
		}
		return fmt.Errorf("GetUnit: %w", err)
	}
	if cfg.ActionsConfig().IsWorkflowDisabled(row.Schedule.WorkflowID) {
		return nil
	}

	if err := CreateScheduleTask(ctx, row); err != nil {
		return fmt.Errorf("create run for %s workflow %q: %w", row.Repo.FullName(), row.Schedule.WorkflowID, err)
	}
	return nil
}

// CreateScheduleTask creates a scheduled task from a cron action schedule spec.
// It creates an action run based on the schedule, inserts it into the database, and creates commit statuses for each job.
func CreateScheduleTask(ctx context.Context, spec *actions_model.ActionScheduleSpec) error {
	cron := spec.Schedule

	// Scheduled runs carry no webhook payload; synthesize what github.event.* expects.
	if err := spec.Repo.LoadOwner(ctx); err != nil {
		return fmt.Errorf("LoadOwner: %w", err)
	}
	fields := map[string]any{
		"repository": convert.ToRepo(ctx, spec.Repo, access_model.Permission{AccessMode: perm_model.AccessModeRead}),
		"sender":     convert.ToUser(ctx, user_model.NewActionsUser(), nil),
	}
	if spec.Repo.Owner.IsOrganization() {
		fields["organization"] = convert.ToOrganization(ctx, organization.OrgFromUser(spec.Repo.Owner))
	}
	eventPayload := withScheduleInEventPayload(cron.EventPayload, spec.Spec, fields)

	// Create a new action run based on the schedule
	run := &actions_model.ActionRun{
		Title:         cron.Title,
		RepoID:        cron.RepoID,
		OwnerID:       cron.OwnerID,
		WorkflowID:    cron.WorkflowID,
		TriggerUserID: cron.TriggerUserID,
		Ref:           cron.Ref,
		CommitSHA:     cron.CommitSHA,
		Event:         cron.Event,
		EventPayload:  eventPayload,
		TriggerEvent:  string(webhook_module.HookEventSchedule),
		ScheduleID:    cron.ID,
		Status:        actions_model.StatusWaiting,
		// schedule runs the repo's own workflow at the recorded commit
		WorkflowRepoID:    cron.RepoID,
		WorkflowCommitSHA: cron.CommitSHA,
	}

	// FIXME cron.Content might be outdated if the workflow file has been changed.
	// Load the latest sha from default branch
	// Insert the action run and its associated jobs into the database
	if err := PrepareRunAndInsert(ctx, cron.Content, run, nil); err != nil {
		return err
	}

	// Return nil if no errors occurred
	return nil
}

func withScheduleInEventPayload(eventPayload, schedule string, fields map[string]any) string {
	if schedule == "" {
		return eventPayload
	}

	// eventPayload originates from json.Marshal(input.Payload) in handleSchedules,
	// so a nil payload is stored as the literal "null" and pre-existing rows may be
	// empty. Both cases start from a fresh map so the schedule field can still be set.
	var event map[string]any
	if eventPayload != "" {
		if err := json.Unmarshal([]byte(eventPayload), &event); err != nil {
			log.Error("withScheduleInEventPayload: unmarshal: %v", err)
			return eventPayload
		}
	}
	if event == nil {
		event = map[string]any{}
	}

	maps.Copy(event, fields)
	event["schedule"] = schedule
	updatedPayload, err := json.Marshal(event)
	if err != nil {
		log.Error("withScheduleInEventPayload: marshal: %v", err)
		return eventPayload
	}

	return string(updatedPayload)
}