1package jobs
2
3import (
4 "context"
5 "fmt"
6 "path/filepath"
7 "runtime"
8 "strings"
9
10 log "github.com/charmbracelet/log/v2"
11 "github.com/charmbracelet/soft-serve/git"
12 "github.com/charmbracelet/soft-serve/pkg/backend"
13 "github.com/charmbracelet/soft-serve/pkg/config"
14 "github.com/charmbracelet/soft-serve/pkg/db"
15 "github.com/charmbracelet/soft-serve/pkg/lfs"
16 "github.com/charmbracelet/soft-serve/pkg/store"
17 "github.com/charmbracelet/soft-serve/pkg/sync"
18)
19
20func init() {
21 Register("mirror-pull", mirrorPull{})
22}
23
24type mirrorPull struct{}
25
26// Spec derives the spec used for pull mirrors and implements Runner.
27func (m mirrorPull) Spec(ctx context.Context) string {
28 cfg := config.FromContext(ctx)
29 if cfg.Jobs.MirrorPull != "" {
30 return cfg.Jobs.MirrorPull
31 }
32 return "@every 10m"
33}
34
35// Func runs the (pull) mirror job task and implements Runner.
36func (m mirrorPull) Func(ctx context.Context) func() {
37 cfg := config.FromContext(ctx)
38 logger := log.FromContext(ctx).WithPrefix("jobs.mirror")
39 b := backend.FromContext(ctx)
40 dbx := db.FromContext(ctx)
41 datastore := store.FromContext(ctx)
42 return func() {
43 repos, err := b.Repositories(ctx)
44 if err != nil {
45 logger.Error("error getting repositories", "err", err)
46 return
47 }
48
49 // Divide the work up among the number of CPUs.
50 wq := sync.NewWorkPool(ctx, runtime.GOMAXPROCS(0),
51 sync.WithWorkPoolLogger(logger.Errorf),
52 )
53
54 logger.Debug("updating mirror repos")
55 for _, repo := range repos {
56 if repo.IsMirror() {
57 r, err := repo.Open()
58 if err != nil {
59 logger.Error("error opening repository", "repo", repo.Name(), "err", err)
60 continue
61 }
62
63 name := repo.Name()
64 wq.Add(name, func() {
65 repo := repo
66
67 cmds := []string{
68 "fetch --prune", // fetch prune before updating remote
69 "remote update --prune", // update remote and prune remote refs
70 }
71
72 for _, c := range cmds {
73 args := strings.Split(c, " ")
74 cmd := git.NewCommand(args...).WithContext(ctx)
75 cmd.AddEnvs(
76 fmt.Sprintf(`GIT_SSH_COMMAND=ssh -o UserKnownHostsFile="%s" -o StrictHostKeyChecking=no -i "%s"`,
77 filepath.Join(cfg.DataPath, "ssh", "known_hosts"),
78 cfg.SSH.ClientKeyPath,
79 ),
80 )
81
82 if _, err := cmd.RunInDir(r.Path); err != nil {
83 logger.Error("error running git remote update", "repo", name, "err", err)
84 }
85 }
86
87 if cfg.LFS.Enabled {
88 rcfg, err := r.Config()
89 if err != nil {
90 logger.Error("error getting git config", "repo", name, "err", err)
91 return
92 }
93
94 lfsEndpoint := rcfg.Section("lfs").Option("url")
95 if lfsEndpoint == "" {
96 // If there is no LFS url defined, means the repo
97 // doesn't use LFS and we can skip it.
98 return
99 }
100
101 ep, err := lfs.NewEndpoint(lfsEndpoint)
102 if err != nil {
103 logger.Error("error creating LFS endpoint", "repo", name, "err", err)
104 return
105 }
106
107 client := lfs.NewClient(ep)
108 if client == nil {
109 logger.Errorf("failed to create lfs client: unsupported endpoint %s", lfsEndpoint)
110 return
111 }
112
113 if err := backend.StoreRepoMissingLFSObjects(ctx, repo, dbx, datastore, client); err != nil {
114 logger.Error("failed to store missing lfs objects", "err", err, "path", r.Path)
115 return
116 }
117 }
118 })
119 }
120 }
121
122 wq.Run()
123 }
124}