GitRoot

  1// SPDX-FileCopyrightText: 2026 Romain Maneschi <romain@gitroot.dev>
  2//
  3// SPDX-License-Identifier: EUPL-1.2
  4
  5package scheduler
  6
  7import (
  8	"context"
  9	"slices"
 10	"sync"
 11	"time"
 12
 13	"github.com/adhocore/gronx"
 14	"github.com/samber/oops"
 15	"gitroot.dev/server/background"
 16	"gitroot.dev/server/logger"
 17	"gitroot.dev/server/repository"
 18)
 19
 20type Manager struct {
 21	logger            *logger.Logger
 22	conf              needConf
 23	repoManager       needRepoManager
 24	backgroundManager needBackgroundManager
 25	tasksLock         sync.RWMutex
 26	tasks             []*task
 27}
 28
 29type needConf interface {
 30}
 31
 32type needRepoManager interface {
 33	GetRepositories(ctx context.Context) ([]string, error)
 34	Open(ctx context.Context, repoName string) (*repository.GitRootRepository, error)
 35}
 36
 37type needBackgroundManager interface {
 38	Schedule(task background.ScheduleTaskInput)
 39}
 40
 41func NewManager(conf needConf, repoManager needRepoManager) *Manager {
 42	m := &Manager{
 43		logger:      logger.NewLogger(logger.SCHEDULER_MANAGER),
 44		conf:        conf,
 45		repoManager: repoManager,
 46	}
 47	return m
 48}
 49
 50func (m *Manager) SetBackgroundManager(backgroundManager needBackgroundManager) {
 51	m.backgroundManager = backgroundManager
 52}
 53
 54func (m *Manager) findSchedule(ctx context.Context) ([]*task, error) {
 55	repositories, err := m.repoManager.GetRepositories(ctx)
 56	if err != nil {
 57		return nil, oops.Wrapf(err, "can't get repositories")
 58	}
 59	allSchedule := []*task{}
 60	for _, r := range repositories {
 61		all, err := m.findScheduleInProject(ctx, r)
 62		if err != nil {
 63			m.logger.Error("can't find tasks in project", err, logger.NewLoggerPair("project", r))
 64		}
 65		allSchedule = append(allSchedule, all...)
 66	}
 67	return allSchedule, nil
 68}
 69
 70func (m *Manager) Start(ctx context.Context) {
 71	allSchedule, err := m.findSchedule(ctx)
 72	if err != nil {
 73		m.logger.Error("can't find schedule", err)
 74	}
 75	m.tasksLock.Lock()
 76	m.tasks = allSchedule
 77	m.tasksLock.Unlock()
 78
 79	ticker := time.NewTicker(1 * time.Second)
 80out:
 81	for {
 82		select {
 83		case <-ticker.C:
 84			m.tick(time.Now())
 85		case <-ctx.Done():
 86			m.logger.Debug("scheduler off")
 87			break out
 88		}
 89	}
 90}
 91
 92func (m *Manager) tick(now time.Time) {
 93	m.tasksLock.RLock()
 94	defer m.tasksLock.RUnlock()
 95
 96	for _, task := range m.tasks {
 97		if !now.Before(task.nextRun) {
 98			m.logger.Debug("before next tick", logger.NewLoggerPair("nextRun", task.nextRun), logger.NewLoggerPair("expr", task.cronExpr), logger.NewLoggerPair("repoName", task.repoName), logger.NewLoggerPair("plugin", task.pluginPurl))
 99			nextTime, err := gronx.NextTickAfter(task.cronExpr, task.nextRun, false)
100			if err != nil {
101				m.logger.Error("invalid cron expr in tick", err, logger.NewLoggerPair("expr", task.cronExpr), logger.NewLoggerPair("repoName", task.repoName), logger.NewLoggerPair("plugin", task.pluginPurl))
102				continue
103			}
104			task.nextRun = nextTime
105			m.logger.Debug("after next tick", logger.NewLoggerPair("nextRun", task.nextRun), logger.NewLoggerPair("expr", task.cronExpr), logger.NewLoggerPair("repoName", task.repoName), logger.NewLoggerPair("plugin", task.pluginPurl))
106
107			m.backgroundManager.Schedule(background.ScheduleTaskInput{
108				RepoName:   task.repoName,
109				PluginPurl: task.pluginPurl,
110				PluginRun:  task.pluginRun,
111			})
112		}
113	}
114}
115
116func (m *Manager) LoadChanges(ctx context.Context, repoName string) {
117	m.logger.Info("reload tasks", logger.NewLoggerPair("repo", repoName))
118	all, err := m.findScheduleInProject(ctx, repoName)
119	if err != nil {
120		m.logger.Error("can't find tasks in project", err, logger.NewLoggerPair("project", repoName))
121	}
122	m.logger.Info("find tasks", logger.NewLoggerPair("repo", repoName), logger.NewLoggerPair("nb", len(all)))
123	m.tasksLock.Lock()
124	defer m.tasksLock.Unlock()
125	m.tasks = slices.DeleteFunc(m.tasks, func(t *task) bool {
126		return t.repoName == repoName
127	})
128	m.tasks = append(m.tasks, all...)
129	m.logger.Info("finish LoadChanges", logger.NewLoggerPair("repo", repoName))
130}