Federated via graft
12 kB
310 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
68func (g *GitSyncer) run(args ...string) (string, error) {
69 ctx, cancel := context.WithTimeout(context.Background(), execTimeout)
70 defer cancel()
71 cmd := exec.CommandContext(ctx, "git", args...)
72 cmd.Dir = g.WorkDir
73 cmd.Env = append(os.Environ(), "RAD_HOME="+g.RadHome)
74 var out bytes.Buffer
75 cmd.Stdout = &out
76 cmd.Stderr = &out
77 err := cmd.Run()
78 if err != nil {
79 return out.String(), fmt.Errorf("git %v: %w: %s", args, err, out.String())
80 }
81 return out.String(), nil
82}
83
84// ensureClone creates the working clone and its two remotes if this is the
85// first run for this repo pair. The rad remote mirrors exactly what
86// `rad init` itself sets up (verified against a live repo): a fetch URL of
87// rad://<rid>, and a separate push URL of rad://<rid>/<local-node-id> —
88// git-remote-rad needs the pushing node's own id in the push URL.
89func (g *GitSyncer) ensureClone() error {
90 if _, err := os.Stat(filepath.Join(g.WorkDir, ".git")); err == nil {
91 return nil
92 }
93 if err := os.MkdirAll(g.WorkDir, 0o700); err != nil {
94 return fmt.Errorf("create work dir: %w", err)
95 }
96 if _, err := g.run("init", "-q", "-b", g.ForgejoBranch); err != nil {
97 return err
98 }
99 if _, err := g.run("remote", "add", "forgejo", g.ForgejoURL); err != nil {
100 return err
101 }
102
103 ridBare := strings.TrimPrefix(g.RID, "rad:")
104 nid, err := g.localNodeID()
105 if err != nil {
106 return fmt.Errorf("resolve local node id: %w", err)
107 }
108 if _, err := g.run("remote", "add", "rad", "rad://"+ridBare); err != nil {
109 return err
110 }
111 if _, err := g.run("remote", "set-url", "--push", "rad", "rad://"+ridBare+"/"+nid); err != nil {
112 return err
113 }
114 return nil
115}
116
117// localNodeID returns the Radicle node id (DID, without the did:key:
118// prefix) that owns g.RadHome.
119func (g *GitSyncer) localNodeID() (string, error) {
120 // --nid is deprecated in favor of `rad node status --only nid` and
121 // prints a warning to stderr; keep that separate from stdout so it
122 // never ends up concatenated into the id itself.
123 ctx, cancel := context.WithTimeout(context.Background(), execTimeout)
124 defer cancel()
125 cmd := exec.CommandContext(ctx, "rad", "self", "--nid")
126 cmd.Env = append(os.Environ(), "RAD_HOME="+g.RadHome)
127 var out, errOut bytes.Buffer
128 cmd.Stdout = &out
129 cmd.Stderr = &errOut
130 if err := cmd.Run(); err != nil {
131 return "", fmt.Errorf("rad self --nid: %w: %s", err, errOut.String())
132 }
133 return strings.TrimSpace(out.String()), nil
134}
135
136// headOf returns the commit SHA a remote branch currently points to, or ""
137// if the branch does not exist yet on that remote.
138func (g *GitSyncer) headOf(remote, branch string) (string, error) {
139 out, err := g.run("ls-remote", remote, "refs/heads/"+branch)
140 if err != nil {
141 return "", err
142 }
143 if out == "" {
144 return "", nil
145 }
146 var sha string
147 fmt.Sscanf(out, "%s", &sha)
148 return sha, nil
149}
150
151// isAncestor reports whether commit a is an ancestor of commit b (i.e.
152// fast-forwarding from a to b is safe).
153func (g *GitSyncer) isAncestor(a, b string) bool {
154 _, err := g.run("merge-base", "--is-ancestor", a, b)
155 return err == nil
156}
157
158// Sync mirrors the default branch in whichever direction has new,
159// fast-forwardable commits. It never force-pushes: a genuine divergence
160// (commits on both sides since the last sync, neither an ancestor of the
161// other) is reported as an error so the operator can resolve it by hand.
162func (g *GitSyncer) Sync() error {
163 if err := g.ensureClone(); err != nil {
164 return err
165 }
166
167 forgejoHead, err := g.headOf("forgejo", g.ForgejoBranch)
168 if err != nil {
169 return fmt.Errorf("read forgejo head: %w", err)
170 }
171 radHead, err := g.headOf("rad", g.ForgejoBranch)
172 if err != nil {
173 return fmt.Errorf("read rad head: %w", err)
174 }
175
176 lastForgejo, lastRad, err := g.State.GitCursor(g.RepoPair)
177 if err != nil {
178 return fmt.Errorf("read cursor: %w", err)
179 }
180
181 switch {
182 case forgejoHead == radHead:
183 // Already in sync; still record the cursor in case this is the
184 // first run and it was empty before.
185 g.aiSkipStreak = 0
186
187 case radHead == "" || radHead == lastRad:
188 // Radicle hasn't moved since we last looked: forgejo is ahead.
189 if err := g.pushBranch("forgejo", "rad", radHead, state.ForgejoToRadicle); err != nil {
190 return fmt.Errorf("mirror forgejo -> rad: %w", err)
191 }
192
193 case forgejoHead == "" || forgejoHead == lastForgejo:
194 // Forgejo hasn't moved: rad is ahead.
195 if g.AuthorizedIntegrationSource != "" && g.aiSkipStreak < maxAISkipStreak {
196 // Someone else's CI already pushes directly into this
197 // Forgejo (see AuthorizedIntegrationSource's doc comment) —
198 // faster than waiting for us to relay it via Radicle.
199 // Nothing to do; next pass sees forgejoHead == radHead once
200 // it arrives, same as any other convergence.
201 g.aiSkipStreak++
202 break
203 }
204 if err := g.pushBranch("rad", "forgejo", forgejoHead, state.RadicleToForgejo); err != nil {
205 return fmt.Errorf("mirror rad -> forgejo: %w", err)
206 }
207
208 case g.isAncestor(lastForgejo, forgejoHead) && g.isAncestor(lastRad, radHead) &&
209 g.isAncestor(radHead, forgejoHead):
210 // Both moved, but rad's tip is contained in forgejo's: forgejo wins.
211 if err := g.pushBranch("forgejo", "rad", radHead, state.ForgejoToRadicle); err != nil {
212 return fmt.Errorf("mirror forgejo -> rad: %w", err)
213 }
214
215 case g.isAncestor(forgejoHead, radHead):
216 // Symmetric case: forgejo's tip is contained in rad's.
217 if g.AuthorizedIntegrationSource != "" && g.aiSkipStreak < maxAISkipStreak {
218 g.aiSkipStreak++
219 break
220 }
221 if err := g.pushBranch("rad", "forgejo", forgejoHead, state.RadicleToForgejo); err != nil {
222 return fmt.Errorf("mirror rad -> forgejo: %w", err)
223 }
224
225 default:
226 return fmt.Errorf(
227 "%s: forgejo and rad diverged since last sync (forgejo=%s rad=%s, last forgejo=%s rad=%s); resolve manually, not auto-merging",
228 g.RepoPair, forgejoHead, radHead, lastForgejo, lastRad)
229 }
230
231 forgejoHead, err = g.headOf("forgejo", g.ForgejoBranch)
232 if err != nil {
233 return err
234 }
235 radHead, err = g.headOf("rad", g.ForgejoBranch)
236 if err != nil {
237 return err
238 }
239 return g.State.SetGitCursor(g.RepoPair, forgejoHead, radHead)
240}
241
242// pushBranch fetches g.ForgejoBranch from "from" and pushes it to "to".
243// oldToHead is that branch's head on "to" before the push (possibly ""),
244// used only to log which commits this pass actually moved, for the status
245// page's activity calendar.
246func (g *GitSyncer) pushBranch(from, to, oldToHead, direction string) error {
247 if _, err := g.run("fetch", "-q", from, g.ForgejoBranch); err != nil {
248 return fmt.Errorf("fetch %s: %w", from, err)
249 }
250 if _, err := g.run("push", to, "FETCH_HEAD:refs/heads/"+g.ForgejoBranch); err != nil {
251 return fmt.Errorf("push %s: %w", to, err)
252 }
253 g.logMirroredCommits(oldToHead, direction, to)
254 return nil
255}
256
257// logMirroredCommits records one activity_log entry per commit newly
258// reachable from FETCH_HEAD that wasn't reachable from oldHead. Best-effort:
259// a failure here never fails the sync itself, it just leaves the status
260// page's calendar short an entry. to is "forgejo" or "rad" — the remote the
261// commits were just pushed to — so the logged link (and the dashboard's
262// derived server label) points at where the content actually landed,
263// instead of always pointing at Forgejo regardless of direction.
264func (g *GitSyncer) logMirroredCommits(oldHead, direction, to string) {
265 rangeSpec := "FETCH_HEAD"
266 if oldHead != "" {
267 rangeSpec = oldHead + "..FETCH_HEAD"
268 }
269 // %x1f (unit separator) between fields: commit subjects can contain
270 // anything else printable, so a space or comma delimiter would be
271 // ambiguous to split back apart.
272 out, err := g.run("log", "--pretty=format:%H%x1f%h %s", rangeSpec)
273 if err != nil || out == "" {
274 return
275 }
276 for _, line := range strings.Split(out, "\n") {
277 if line == "" {
278 continue
279 }
280 full, summary, ok := strings.Cut(line, "\x1f")
281 if !ok {
282 continue
283 }
284 url := ""
285 switch to {
286 case "rad":
287 if g.RadicleWebURL != "" {
288 url = g.RadicleWebURL + "/commits/" + full
289 }
290 default:
291 if g.ForgejoWebURL != "" {
292 url = g.ForgejoWebURL + "/commit/" + full
293 }
294 }
295 id, _ := g.State.LogActivity(state.Activity{
296 RepoPair: g.RepoPair, Series: g.Series, SeriesURL: g.SeriesURL,
297 Kind: "git", Direction: direction, Summary: summary, URL: url,
298 })
299 // Best-effort, same as LogActivity above: a Bluesky hiccup never
300 // fails the sync, and nil Bluesky (the common case, no account
301 // configured) is a silent no-op.
302 text := g.Series + ": " + summary
303 if url != "" {
304 text += "\n" + url
305 }
306 if postURI, err := g.Bluesky.Post(text); err == nil && postURI != "" {
307 g.State.SaveATProtoPost(postURI, id, g.Series)
308 }
309 }
310}