Federated via graft
6.2 kB
235 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 pairsBySeries := map[string][]*sync.RepoSyncer{}
53 for _, pair := range cfg.Repos {
54 rs, err := sync.New(pair, st, sync.WorkDirFor(stateDir, pair.Name))
55 if err != nil {
56 log.Error("set up repo pair", "pair", pair.Name, "err", err)
57 os.Exit(1)
58 }
59 syncers = append(syncers, rs)
60 syncersByPair[pair.Name] = rs
61 if _, ok := syncersBySeries[rs.Series()]; !ok {
62 syncersBySeries[rs.Series()] = rs
63 }
64 pairsBySeries[rs.Series()] = append(pairsBySeries[rs.Series()], rs)
65 }
66 detectAuthorizedIntegrations(pairsBySeries, log)
67
68 topology := buildTopology(cfg)
69 blueskyConfigured := map[string]bool{}
70 for _, pair := range cfg.Repos {
71 if pair.Bluesky != nil {
72 series := pair.Series
73 if series == "" {
74 series = pair.Name
75 }
76 blueskyConfigured[series] = true
77 }
78 }
79
80 tracker := status.NewTracker(st, cfg.SourceURL)
81 tracker.SetTopology(topology)
82 tracker.SetPublicHost(cfg.PublicHost)
83 tracker.SetBlueskyConfigured(blueskyConfigured)
84
85 var apHandler *activitypub.Handler
86 if cfg.PublicHost != "" {
87 apHandler = activitypub.NewHandler(st, cfg.PublicHost, log,
88 func(series string) string {
89 if refs := topology[series]; len(refs) > 0 {
90 return refs[0].URL
91 }
92 return ""
93 },
94 func(series string) bool {
95 _, ok := topology[series]
96 return ok
97 },
98 func(repoPair, kind string, forgejoID int64, radicleID, body string) error {
99 rs, ok := syncersByPair[repoPair]
100 if !ok {
101 return fmt.Errorf("unknown repo pair %q", repoPair)
102 }
103 return rs.CommentOnItem(kind, forgejoID, radicleID, body)
104 },
105 func(series string) string {
106 rs, ok := syncersBySeries[series]
107 if !ok {
108 return ""
109 }
110 did, err := rs.RadicleDID()
111 if err != nil {
112 log.Error("resolve radicle did", "series", series, "err", err)
113 return ""
114 }
115 return did
116 },
117 )
118 }
119
120 runAll := func() {
121 for _, rs := range syncers {
122 gitErr, issuesErr, patchErr := rs.Run(log)
123 tracker.Record(rs.Name(), rs.Series(), gitErr, issuesErr, patchErr)
124 }
125 if apHandler != nil {
126 apHandler.DeliverNewActivity()
127 }
128 }
129
130 if *listen != "" {
131 mux := http.NewServeMux()
132 if apHandler != nil {
133 apHandler.Register(mux)
134 }
135 mux.Handle("/", tracker.Handler())
136
137 go func() {
138 if err := http.ListenAndServe(*listen, mux); err != nil {
139 log.Error("status server stopped", "err", err)
140 }
141 }()
142 }
143
144 runAll()
145 if *once {
146 return
147 }
148
149 ticker := time.NewTicker(cfg.SyncInterval)
150 defer ticker.Stop()
151 for range ticker.C {
152 runAll()
153 }
154}
155
156// buildTopology declares every mirror target configured for each series
157// (its Forgejo side and its Radicle side), so the dashboard can show a
158// side that's never produced an event yet — not just sides the activity
159// log happens to mention.
160func buildTopology(cfg *config.Config) map[string][]status.ServerRef {
161 topology := map[string][]status.ServerRef{}
162 seen := map[string]bool{}
163 add := func(series, host, repoURL string) {
164 if host == "" {
165 return
166 }
167 key := series + "|" + host
168 if seen[key] {
169 return
170 }
171 seen[key] = true
172 topology[series] = append(topology[series], status.ServerRef{Label: host, URL: repoURL})
173 }
174
175 for _, pair := range cfg.Repos {
176 series := pair.Series
177 if series == "" {
178 series = pair.Name
179 }
180
181 fHost := ""
182 if u, err := url.Parse(pair.Forgejo.BaseURL); err == nil {
183 fHost = u.Host
184 }
185 fURL := strings.TrimRight(pair.Forgejo.BaseURL, "/") + "/" + pair.Forgejo.Owner + "/" + pair.Forgejo.Repo
186 add(series, fHost, fURL)
187
188 rHost := ""
189 if u, err := url.Parse(pair.Radicle.HTTPBaseURL); err == nil {
190 rHost = u.Host
191 }
192 add(series, rHost, sync.RadicleExplorerLink(pair.Radicle))
193 }
194 return topology
195}
196
197// detectAuthorizedIntegrations checks, for every pair of Forgejo hosts
198// sharing a series, whether one side already has a Forgejo Actions
199// workflow pushing directly into the other via Authorized Integrations
200// (see internal/sync's authint.go) — and if so, tells the receiving
201// pair's GitSyncer to stop relaying that content via Radicle itself,
202// since the direct path is faster. Run once at startup: workflow files
203// don't change often enough to justify re-checking every pass.
204func detectAuthorizedIntegrations(pairsBySeries map[string][]*sync.RepoSyncer, log *slog.Logger) {
205 for series, pairs := range pairsBySeries {
206 if len(pairs) < 2 {
207 continue
208 }
209 for _, source := range pairs {
210 for _, dest := range pairs {
211 if source == dest {
212 continue
213 }
214 destHost := dest.ForgejoHost()
215 if destHost == "" {
216 continue
217 }
218 if sync.HasAuthorizedIntegration(source.ForgejoClient(), destHost) {
219 log.Info("authorized integration detected",
220 "series", series, "source", source.ForgejoHost(), "target", destHost)
221 dest.SetAuthorizedIntegrationSource(source.ForgejoHost())
222 }
223 }
224 }
225 }
226}
227
228func lastSlash(s string) int {
229 for i := len(s) - 1; i >= 0; i-- {
230 if s[i] == '/' {
231 return i
232 }
233 }
234 return -1
235}