Federated via graft
12 kB
316 lines
1// Package sync is the actual synchronization brick: it takes a Forgejo
2// client and a Radicle client for one repo pair and mirrors content between
3// them, using the state store to stay idempotent across runs.
4//
5// Scope of this first version, deliberately: fast-forward-only git mirroring
6// (a real divergence is logged and skipped, never force-pushed), and
7// create-only issue/patch mirroring (edits made after the initial mirror are
8// not propagated — that needs per-item diffing, left for a follow-up once
9// the create path is proven in practice).
10package sync
11
12import (
13 "bytes"
14 "context"
15 "fmt"
16 "os"
17 "os/exec"
18 "path/filepath"
19 "strings"
20 "time"
21
22 "graft/internal/state"
23)
24
25// execTimeout bounds every git/rad subprocess this package runs. Without
26// it, one hung fetch/push (a slow network, an unresponsive Radicle node)
27// would block the single-threaded sync loop forever — every other
28// configured pair too, not just the stuck one.
29const execTimeout = 5 * time.Minute
30
31// GitSyncer mirrors a repository's default branch between Forgejo and
32// Radicle using a local working clone with two remotes.
33type GitSyncer struct {
34 RepoPair string // key used in the state store
35 WorkDir string // local clone, created on first run if absent
36 ForgejoURL string // authenticated clone URL (token embedded), for git
37 ForgejoWebURL string // https://host/owner/repo, for links in the UI
38 RadicleWebURL string // https://explorer/nodes/host/rid, for links in the UI
39 ForgejoBranch string
40 RID string
41 RadHome string
42 State *state.Store
43 Series string // dashboard row this pair's activity groups under
44 SeriesURL string // where clicking that row's name goes
45 Bluesky *BlueskyPoster // nil unless this pair has bluesky configured
46 // AuthorizedIntegrationSource, if set, names another pair's Forgejo
47 // host that already pushes new commits directly into this pair's
48 // Forgejo via Authorized Integrations (see authint.go) — faster than
49 // this pair waiting for the next poll to relay it via Radicle. When
50 // set, Sync skips its own rad -> forgejo push and just observes
51 // whether the content has already arrived, falling back to pushing it
52 // itself only once a real divergence would otherwise be reported —
53 // or after maxAISkipStreak consecutive passes without convergence
54 // (see Sync), since detection here is a content heuristic (authint.go),
55 // not a guarantee the workflow is actually delivering.
56 AuthorizedIntegrationSource string
57 aiSkipStreak int
58}
59
60// maxAISkipStreak bounds how many consecutive passes GitSyncer will trust
61// AuthorizedIntegrationSource without seeing forgejo actually catch up to
62// radicle. Detection is a heuristic (a matching string in a workflow
63// file), not proof the integration is configured correctly or even still
64// working — without this bound, a stale or misconfigured workflow would
65// make graft silently stop mirroring that direction forever.
66const maxAISkipStreak = 10
67
68// run captures stdout and stderr separately — a git warning (e.g. "warning:
69// redirecting to ..." on a renamed repo) is printed to stderr and must
70// never end up in the stdout headOf parses for a commit SHA. Discovered via
71// a real graft-presentation pair whose Forgejo peer got renamed: git's
72// redirect warning landed first in the merged buffer and headOf's Sscanf
73// picked "warning:" up as the SHA instead of skipping to the real line.
74func (g *GitSyncer) run(args ...string) (string, error) {
75 ctx, cancel := context.WithTimeout(context.Background(), execTimeout)
76 defer cancel()
77 cmd := exec.CommandContext(ctx, "git", args...)
78 cmd.Dir = g.WorkDir
79 cmd.Env = append(os.Environ(), "RAD_HOME="+g.RadHome)
80 var out, errOut bytes.Buffer
81 cmd.Stdout = &out
82 cmd.Stderr = &errOut
83 err := cmd.Run()
84 if err != nil {
85 return out.String(), fmt.Errorf("git %v: %w: %s", args, err, errOut.String())
86 }
87 return out.String(), nil
88}
89
90// ensureClone creates the working clone and its two remotes if this is the
91// first run for this repo pair. The rad remote mirrors exactly what
92// `rad init` itself sets up (verified against a live repo): a fetch URL of
93// rad://<rid>, and a separate push URL of rad://<rid>/<local-node-id> —
94// git-remote-rad needs the pushing node's own id in the push URL.
95func (g *GitSyncer) ensureClone() error {
96 if _, err := os.Stat(filepath.Join(g.WorkDir, ".git")); err == nil {
97 return nil
98 }
99 if err := os.MkdirAll(g.WorkDir, 0o700); err != nil {
100 return fmt.Errorf("create work dir: %w", err)
101 }
102 if _, err := g.run("init", "-q", "-b", g.ForgejoBranch); err != nil {
103 return err
104 }
105 if _, err := g.run("remote", "add", "forgejo", g.ForgejoURL); err != nil {
106 return err
107 }
108
109 ridBare := strings.TrimPrefix(g.RID, "rad:")
110 nid, err := g.localNodeID()
111 if err != nil {
112 return fmt.Errorf("resolve local node id: %w", err)
113 }
114 if _, err := g.run("remote", "add", "rad", "rad://"+ridBare); err != nil {
115 return err
116 }
117 if _, err := g.run("remote", "set-url", "--push", "rad", "rad://"+ridBare+"/"+nid); err != nil {
118 return err
119 }
120 return nil
121}
122
123// localNodeID returns the Radicle node id (DID, without the did:key:
124// prefix) that owns g.RadHome.
125func (g *GitSyncer) localNodeID() (string, error) {
126 // --nid is deprecated in favor of `rad node status --only nid` and
127 // prints a warning to stderr; keep that separate from stdout so it
128 // never ends up concatenated into the id itself.
129 ctx, cancel := context.WithTimeout(context.Background(), execTimeout)
130 defer cancel()
131 cmd := exec.CommandContext(ctx, "rad", "self", "--nid")
132 cmd.Env = append(os.Environ(), "RAD_HOME="+g.RadHome)
133 var out, errOut bytes.Buffer
134 cmd.Stdout = &out
135 cmd.Stderr = &errOut
136 if err := cmd.Run(); err != nil {
137 return "", fmt.Errorf("rad self --nid: %w: %s", err, errOut.String())
138 }
139 return strings.TrimSpace(out.String()), nil
140}
141
142// headOf returns the commit SHA a remote branch currently points to, or ""
143// if the branch does not exist yet on that remote.
144func (g *GitSyncer) headOf(remote, branch string) (string, error) {
145 out, err := g.run("ls-remote", remote, "refs/heads/"+branch)
146 if err != nil {
147 return "", err
148 }
149 if out == "" {
150 return "", nil
151 }
152 var sha string
153 fmt.Sscanf(out, "%s", &sha)
154 return sha, nil
155}
156
157// isAncestor reports whether commit a is an ancestor of commit b (i.e.
158// fast-forwarding from a to b is safe).
159func (g *GitSyncer) isAncestor(a, b string) bool {
160 _, err := g.run("merge-base", "--is-ancestor", a, b)
161 return err == nil
162}
163
164// Sync mirrors the default branch in whichever direction has new,
165// fast-forwardable commits. It never force-pushes: a genuine divergence
166// (commits on both sides since the last sync, neither an ancestor of the
167// other) is reported as an error so the operator can resolve it by hand.
168func (g *GitSyncer) Sync() error {
169 if err := g.ensureClone(); err != nil {
170 return err
171 }
172
173 forgejoHead, err := g.headOf("forgejo", g.ForgejoBranch)
174 if err != nil {
175 return fmt.Errorf("read forgejo head: %w", err)
176 }
177 radHead, err := g.headOf("rad", g.ForgejoBranch)
178 if err != nil {
179 return fmt.Errorf("read rad head: %w", err)
180 }
181
182 lastForgejo, lastRad, err := g.State.GitCursor(g.RepoPair)
183 if err != nil {
184 return fmt.Errorf("read cursor: %w", err)
185 }
186
187 switch {
188 case forgejoHead == radHead:
189 // Already in sync; still record the cursor in case this is the
190 // first run and it was empty before.
191 g.aiSkipStreak = 0
192
193 case radHead == "" || radHead == lastRad:
194 // Radicle hasn't moved since we last looked: forgejo is ahead.
195 if err := g.pushBranch("forgejo", "rad", radHead, state.ForgejoToRadicle); err != nil {
196 return fmt.Errorf("mirror forgejo -> rad: %w", err)
197 }
198
199 case forgejoHead == "" || forgejoHead == lastForgejo:
200 // Forgejo hasn't moved: rad is ahead.
201 if g.AuthorizedIntegrationSource != "" && g.aiSkipStreak < maxAISkipStreak {
202 // Someone else's CI already pushes directly into this
203 // Forgejo (see AuthorizedIntegrationSource's doc comment) —
204 // faster than waiting for us to relay it via Radicle.
205 // Nothing to do; next pass sees forgejoHead == radHead once
206 // it arrives, same as any other convergence.
207 g.aiSkipStreak++
208 break
209 }
210 if err := g.pushBranch("rad", "forgejo", forgejoHead, state.RadicleToForgejo); err != nil {
211 return fmt.Errorf("mirror rad -> forgejo: %w", err)
212 }
213
214 case g.isAncestor(lastForgejo, forgejoHead) && g.isAncestor(lastRad, radHead) &&
215 g.isAncestor(radHead, forgejoHead):
216 // Both moved, but rad's tip is contained in forgejo's: forgejo wins.
217 if err := g.pushBranch("forgejo", "rad", radHead, state.ForgejoToRadicle); err != nil {
218 return fmt.Errorf("mirror forgejo -> rad: %w", err)
219 }
220
221 case g.isAncestor(forgejoHead, radHead):
222 // Symmetric case: forgejo's tip is contained in rad's.
223 if g.AuthorizedIntegrationSource != "" && g.aiSkipStreak < maxAISkipStreak {
224 g.aiSkipStreak++
225 break
226 }
227 if err := g.pushBranch("rad", "forgejo", forgejoHead, state.RadicleToForgejo); err != nil {
228 return fmt.Errorf("mirror rad -> forgejo: %w", err)
229 }
230
231 default:
232 return fmt.Errorf(
233 "%s: forgejo and rad diverged since last sync (forgejo=%s rad=%s, last forgejo=%s rad=%s); resolve manually, not auto-merging",
234 g.RepoPair, forgejoHead, radHead, lastForgejo, lastRad)
235 }
236
237 forgejoHead, err = g.headOf("forgejo", g.ForgejoBranch)
238 if err != nil {
239 return err
240 }
241 radHead, err = g.headOf("rad", g.ForgejoBranch)
242 if err != nil {
243 return err
244 }
245 return g.State.SetGitCursor(g.RepoPair, forgejoHead, radHead)
246}
247
248// pushBranch fetches g.ForgejoBranch from "from" and pushes it to "to".
249// oldToHead is that branch's head on "to" before the push (possibly ""),
250// used only to log which commits this pass actually moved, for the status
251// page's activity calendar.
252func (g *GitSyncer) pushBranch(from, to, oldToHead, direction string) error {
253 if _, err := g.run("fetch", "-q", from, g.ForgejoBranch); err != nil {
254 return fmt.Errorf("fetch %s: %w", from, err)
255 }
256 if _, err := g.run("push", to, "FETCH_HEAD:refs/heads/"+g.ForgejoBranch); err != nil {
257 return fmt.Errorf("push %s: %w", to, err)
258 }
259 g.logMirroredCommits(oldToHead, direction, to)
260 return nil
261}
262
263// logMirroredCommits records one activity_log entry per commit newly
264// reachable from FETCH_HEAD that wasn't reachable from oldHead. Best-effort:
265// a failure here never fails the sync itself, it just leaves the status
266// page's calendar short an entry. to is "forgejo" or "rad" — the remote the
267// commits were just pushed to — so the logged link (and the dashboard's
268// derived server label) points at where the content actually landed,
269// instead of always pointing at Forgejo regardless of direction.
270func (g *GitSyncer) logMirroredCommits(oldHead, direction, to string) {
271 rangeSpec := "FETCH_HEAD"
272 if oldHead != "" {
273 rangeSpec = oldHead + "..FETCH_HEAD"
274 }
275 // %x1f (unit separator) between fields: commit subjects can contain
276 // anything else printable, so a space or comma delimiter would be
277 // ambiguous to split back apart.
278 out, err := g.run("log", "--pretty=format:%H%x1f%h %s", rangeSpec)
279 if err != nil || out == "" {
280 return
281 }
282 for _, line := range strings.Split(out, "\n") {
283 if line == "" {
284 continue
285 }
286 full, summary, ok := strings.Cut(line, "\x1f")
287 if !ok {
288 continue
289 }
290 url := ""
291 switch to {
292 case "rad":
293 if g.RadicleWebURL != "" {
294 url = g.RadicleWebURL + "/commits/" + full
295 }
296 default:
297 if g.ForgejoWebURL != "" {
298 url = g.ForgejoWebURL + "/commit/" + full
299 }
300 }
301 id, _ := g.State.LogActivity(state.Activity{
302 RepoPair: g.RepoPair, Series: g.Series, SeriesURL: g.SeriesURL,
303 Kind: "git", Direction: direction, Summary: summary, URL: url,
304 })
305 // Best-effort, same as LogActivity above: a Bluesky hiccup never
306 // fails the sync, and nil Bluesky (the common case, no account
307 // configured) is a silent no-op.
308 text := g.Series + ": " + summary
309 if url != "" {
310 text += "\n" + url
311 }
312 if postURI, err := g.Bluesky.Post(text); err == nil && postURI != "" {
313 g.State.SaveATProtoPost(postURI, id, g.Series)
314 }
315 }
316}