3a6178315ea6bb83ac2fe73e582aece4204f3cf0

Author
Ayman Bagabas <ayman.bagabas@gmail.com>
Committer
Ayman Bagabas <ayman.bagabas@gmail.com>
Date

Message

refactor: move jobs into its own module

Diff

  1diff --git a/server/cron/cron.go b/server/cron/cron.go
  2index aae506ed54d09edff27558ba2664079e8eafab5f..6fed17b6a15ae6a954e9d6aa8d2418521231b4be 100644
  3--- a/server/cron/cron.go
  4+++ b/server/cron/cron.go
  5@@ -48,3 +48,14 @@ func (s *Scheduler) Shutdown() {
  6 func (s *Scheduler) Start() {
  7 	s.Cron.Start()
  8 }
  9+
 10+// AddFunc adds a job to the Scheduler.
 11+func (s *Scheduler) AddFunc(spec string, fn func()) (int, error) {
 12+	id, err := s.Cron.AddFunc(spec, fn)
 13+	return int(id), err
 14+}
 15+
 16+// Remove removes a job from the Scheduler.
 17+func (s *Scheduler) Remove(id int) {
 18+	s.Cron.Remove(cron.EntryID(id))
 19+}
 20diff --git a/server/jobs/jobs.go b/server/jobs/jobs.go
 21new file mode 100644
 22index 0000000000000000000000000000000000000000..ba39af85c6ae9e632d64d9512f96a9ddccbf27b8
 23--- /dev/null
 24+++ b/server/jobs/jobs.go
 25@@ -0,0 +1,32 @@
 26+package jobs
 27+
 28+import (
 29+	"context"
 30+	"sync"
 31+)
 32+
 33+// Job is a job that can be registered with the scheduler.
 34+type Job struct {
 35+	ID   int
 36+	Spec string
 37+	Func func(context.Context) func()
 38+}
 39+
 40+var (
 41+	mtx  sync.Mutex
 42+	jobs = make(map[string]*Job, 0)
 43+)
 44+
 45+// Register registers a job.
 46+func Register(name, spec string, fn func(context.Context) func()) {
 47+	mtx.Lock()
 48+	defer mtx.Unlock()
 49+	jobs[name] = &Job{Spec: spec, Func: fn}
 50+}
 51+
 52+// List returns a map of registered jobs.
 53+func List() map[string]*Job {
 54+	mtx.Lock()
 55+	defer mtx.Unlock()
 56+	return jobs
 57+}
 58diff --git a/server/jobs.go b/server/jobs/mirror.go
 59rename from server/jobs.go
 60rename to server/jobs/mirror.go
 61index 2cfbd09844200eb92f4596b8f6eb32911775d315..cf6a620a082760ce0bbc4ca64e441c8f611fdce5 100644
 62--- a/server/jobs.go
 63+++ b/server/jobs/mirror.go
 64@@ -1,32 +1,36 @@
 65-package server
 66+package jobs
 67 
 68 import (
 69+	"context"
 70 	"fmt"
 71 	"path/filepath"
 72 	"runtime"
 73 
 74+	"github.com/charmbracelet/log"
 75 	"github.com/charmbracelet/soft-serve/git"
 76 	"github.com/charmbracelet/soft-serve/server/backend"
 77+	"github.com/charmbracelet/soft-serve/server/config"
 78 	"github.com/charmbracelet/soft-serve/server/sync"
 79 )
 80 
 81-var jobSpecs = map[string]string{
 82-	"mirror": "@every 10m",
 83+func init() {
 84+	Register("mirror-pull", "@every 10m", mirrorPull)
 85 }
 86 
 87-// mirrorJob runs the (pull) mirror job task.
 88-func (s *Server) mirrorJob(b *backend.Backend) func() {
 89-	cfg := s.Config
 90-	logger := s.logger
 91+// mirrorPull runs the (pull) mirror job task.
 92+func mirrorPull(ctx context.Context) func() {
 93+	cfg := config.FromContext(ctx)
 94+	logger := log.FromContext(ctx).WithPrefix("jobs.mirror")
 95+	b := backend.FromContext(ctx)
 96 	return func() {
 97-		repos, err := b.Repositories(s.ctx)
 98+		repos, err := b.Repositories(ctx)
 99 		if err != nil {
100 			logger.Error("error getting repositories", "err", err)
101 			return
102 		}
103 
104 		// Divide the work up among the number of CPUs.
105-		wq := sync.NewWorkPool(s.ctx, runtime.GOMAXPROCS(0),
106+		wq := sync.NewWorkPool(ctx, runtime.GOMAXPROCS(0),
107 			sync.WithWorkPoolLogger(logger.Errorf),
108 		)
109 
110@@ -41,7 +45,7 @@ func (s *Server) mirrorJob(b *backend.Backend) func() {
111 
112 				name := repo.Name()
113 				wq.Add(name, func() {
114-					cmd := git.NewCommand("remote", "update", "--prune")
115+					cmd := git.NewCommand("remote", "update", "--prune").WithContext(ctx)
116 					cmd.AddEnvs(
117 						fmt.Sprintf(`GIT_SSH_COMMAND=ssh -o UserKnownHostsFile="%s" -o StrictHostKeyChecking=no -i "%s"`,
118 							filepath.Join(cfg.DataPath, "ssh", "known_hosts"),
119diff --git a/server/server.go b/server/server.go
120index 0b3f23b0e2b3055597d32e4568442218f2a84f92..8c73ad213dd6d68a51f7463604e56d851ff70d47 100644
121--- a/server/server.go
122+++ b/server/server.go
123@@ -13,6 +13,7 @@ import (
124 	"github.com/charmbracelet/soft-serve/server/cron"
125 	"github.com/charmbracelet/soft-serve/server/daemon"
126 	"github.com/charmbracelet/soft-serve/server/db"
127+	"github.com/charmbracelet/soft-serve/server/jobs"
128 	sshsrv "github.com/charmbracelet/soft-serve/server/ssh"
129 	"github.com/charmbracelet/soft-serve/server/stats"
130 	"github.com/charmbracelet/soft-serve/server/web"
131@@ -44,8 +45,8 @@ func NewServer(ctx context.Context) (*Server, error) {
132 	cfg := config.FromContext(ctx)
133 	be := backend.FromContext(ctx)
134 	db := db.FromContext(ctx)
135+	logger := log.FromContext(ctx).WithPrefix("server")
136 	srv := &Server{
137-		Cron:    cron.NewScheduler(ctx),
138 		Config:  cfg,
139 		Backend: be,
140 		DB:      db,
141@@ -54,7 +55,17 @@ func NewServer(ctx context.Context) (*Server, error) {
142 	}
143 
144 	// Add cron jobs.
145-	_, _ = srv.Cron.AddFunc(jobSpecs["mirror"], srv.mirrorJob(be))
146+	sched := cron.NewScheduler(ctx)
147+	for n, j := range jobs.List() {
148+		id, err := sched.AddFunc(j.Spec, j.Func(ctx))
149+		if err != nil {
150+			logger.Warn("error adding cron job", "job", n, "err", err)
151+		}
152+
153+		j.ID = id
154+	}
155+
156+	srv.Cron = sched
157 
158 	srv.SSHServer, err = sshsrv.NewSSHServer(ctx)
159 	if err != nil {
160@@ -133,7 +144,10 @@ func (s *Server) Shutdown(ctx context.Context) error {
161 		return s.StatsServer.Shutdown(ctx)
162 	})
163 	errg.Go(func() error {
164-		s.Cron.Stop()
165+		for _, j := range jobs.List() {
166+			s.Cron.Remove(j.ID)
167+		}
168+		s.Cron.Shutdown()
169 		return nil
170 	})
171 	// defer s.DB.Close() // nolint: errcheck