Federated via graft
4.9 kB
201 lines
1// Command sync runs the Forgejo <-> Radicle sync daemon: on an interval, it
2// mirrors git content, issues and patches for every repo pair listed in the
3// config file, then sleeps until the next pass.
4package main
5
6import (
7 "flag"
8 "fmt"
9 "log/slog"
10 "net/http"
11 "net/url"
12 "os"
13 "strings"
14 "time"
15
16 "graft/internal/activitypub"
17 "graft/internal/config"
18 "graft/internal/state"
19 "graft/internal/status"
20 "graft/internal/sync"
21)
22
23func main() {
24 configPath := flag.String("config", "/etc/graft/config.yaml", "path to config.yaml")
25 once := flag.Bool("once", false, "run a single pass and exit, instead of looping")
26 listen := flag.String("listen", ":8090", "address to serve /healthz and / (status) on; empty disables it")
27 flag.Parse()
28
29 log := slog.New(slog.NewJSONHandler(os.Stdout, nil))
30
31 cfg, err := config.Load(*configPath)
32 if err != nil {
33 log.Error("load config", "err", err)
34 os.Exit(1)
35 }
36
37 st, err := state.Open(cfg.StateDB)
38 if err != nil {
39 log.Error("open state db", "err", err)
40 os.Exit(1)
41 }
42 defer st.Close()
43
44 stateDir := cfg.StateDB
45 if idx := lastSlash(stateDir); idx >= 0 {
46 stateDir = stateDir[:idx]
47 }
48
49 syncers := make([]*sync.RepoSyncer, 0, len(cfg.Repos))
50 syncersByPair := map[string]*sync.RepoSyncer{}
51 syncersBySeries := map[string]*sync.RepoSyncer{}
52 for _, pair := range cfg.Repos {
53 rs, err := sync.New(pair, st, sync.WorkDirFor(stateDir, pair.Name))
54 if err != nil {
55 log.Error("set up repo pair", "pair", pair.Name, "err", err)
56 os.Exit(1)
57 }
58 syncers = append(syncers, rs)
59 syncersByPair[pair.Name] = rs
60 if _, ok := syncersBySeries[rs.Series()]; !ok {
61 syncersBySeries[rs.Series()] = rs
62 }
63 }
64
65 topology := buildTopology(cfg)
66 blueskyConfigured := map[string]bool{}
67 for _, pair := range cfg.Repos {
68 if pair.Bluesky != nil {
69 series := pair.Series
70 if series == "" {
71 series = pair.Name
72 }
73 blueskyConfigured[series] = true
74 }
75 }
76
77 tracker := status.NewTracker(st, cfg.SourceURL)
78 tracker.SetTopology(topology)
79 tracker.SetPublicHost(cfg.PublicHost)
80 tracker.SetBlueskyConfigured(blueskyConfigured)
81
82 var apHandler *activitypub.Handler
83 if cfg.PublicHost != "" {
84 apHandler = activitypub.NewHandler(st, cfg.PublicHost, log,
85 func(series string) string {
86 if refs := topology[series]; len(refs) > 0 {
87 return refs[0].URL
88 }
89 return ""
90 },
91 func(series string) bool {
92 _, ok := topology[series]
93 return ok
94 },
95 func(repoPair, kind string, forgejoID int64, radicleID, body string) error {
96 rs, ok := syncersByPair[repoPair]
97 if !ok {
98 return fmt.Errorf("unknown repo pair %q", repoPair)
99 }
100 return rs.CommentOnItem(kind, forgejoID, radicleID, body)
101 },
102 func(series string) string {
103 rs, ok := syncersBySeries[series]
104 if !ok {
105 return ""
106 }
107 did, err := rs.RadicleDID()
108 if err != nil {
109 log.Error("resolve radicle did", "series", series, "err", err)
110 return ""
111 }
112 return did
113 },
114 )
115 }
116
117 runAll := func() {
118 for _, rs := range syncers {
119 gitErr, issuesErr, patchErr := rs.Run(log)
120 tracker.Record(rs.Name(), rs.Series(), gitErr, issuesErr, patchErr)
121 }
122 if apHandler != nil {
123 apHandler.DeliverNewActivity()
124 }
125 }
126
127 if *listen != "" {
128 mux := http.NewServeMux()
129 if apHandler != nil {
130 apHandler.Register(mux)
131 }
132 mux.Handle("/", tracker.Handler())
133
134 go func() {
135 if err := http.ListenAndServe(*listen, mux); err != nil {
136 log.Error("status server stopped", "err", err)
137 }
138 }()
139 }
140
141 runAll()
142 if *once {
143 return
144 }
145
146 ticker := time.NewTicker(cfg.SyncInterval)
147 defer ticker.Stop()
148 for range ticker.C {
149 runAll()
150 }
151}
152
153// buildTopology declares every mirror target configured for each series
154// (its Forgejo side and its Radicle side), so the dashboard can show a
155// side that's never produced an event yet — not just sides the activity
156// log happens to mention.
157func buildTopology(cfg *config.Config) map[string][]status.ServerRef {
158 topology := map[string][]status.ServerRef{}
159 seen := map[string]bool{}
160 add := func(series, host, repoURL string) {
161 if host == "" {
162 return
163 }
164 key := series + "|" + host
165 if seen[key] {
166 return
167 }
168 seen[key] = true
169 topology[series] = append(topology[series], status.ServerRef{Label: host, URL: repoURL})
170 }
171
172 for _, pair := range cfg.Repos {
173 series := pair.Series
174 if series == "" {
175 series = pair.Name
176 }
177
178 fHost := ""
179 if u, err := url.Parse(pair.Forgejo.BaseURL); err == nil {
180 fHost = u.Host
181 }
182 fURL := strings.TrimRight(pair.Forgejo.BaseURL, "/") + "/" + pair.Forgejo.Owner + "/" + pair.Forgejo.Repo
183 add(series, fHost, fURL)
184
185 rHost := ""
186 if u, err := url.Parse(pair.Radicle.HTTPBaseURL); err == nil {
187 rHost = u.Host
188 }
189 add(series, rHost, sync.RadicleExplorerLink(pair.Radicle))
190 }
191 return topology
192}
193
194func lastSlash(s string) int {
195 for i := len(s) - 1; i >= 0; i-- {
196 if s[i] == '/' {
197 return i
198 }
199 }
200 return -1
201}