1package userbase
2
3import (
4 "context"
5 "encoding/json"
6 "errors"
7 "io"
8 "log/slog"
9 "math/rand/v2"
10 "path/filepath"
11 "strings"
12 "sync"
13 "time"
14
15 "filippo.io/age"
16 dexstorage "github.com/dexidp/dex/storage"
17 "github.com/go-git/go-git/v5"
18 "github.com/go-git/go-git/v5/plumbing"
19 "github.com/go-git/go-git/v5/plumbing/object"
20 "github.com/go-git/go-git/v5/plumbing/transport"
21
22 "tug/internal/platform/replica"
23 "tug/internal/storage"
24 "tug/internal/storage/dexstore"
25)
26
27const RepoDir = "userbase.git"
28
29var errReplicatorConfig = errors.New("userbase: the replicator needs a store, a state directory, a repo and the three keys")
30
31type Keys struct {
32 Identity *age.X25519Identity
33 BreakGlass *age.X25519Recipient
34 Signer *replica.Signer
35 DeployKey []byte
36 KnownHosts string
37}
38
39func (k Keys) Remote(url string) (replica.Remote, error) {
40 if !strings.HasPrefix(url, "ssh://") {
41 return replica.Remote{URL: url}, nil
42 }
43 auth, err := replica.SSHAuth(url, k.DeployKey, k.KnownHosts)
44 if err != nil {
45 return replica.Remote{URL: url}, err
46 }
47 return replica.Remote{URL: url, Auth: auth}, nil
48}
49
50func (k Keys) Digest(blob []byte) ([32]byte, error) {
51 plaintext, err := Open(blob, k.Identity)
52 if err != nil {
53 return [32]byte{}, err
54 }
55 doc, err := Decode(plaintext)
56 if err != nil {
57 return [32]byte{}, err
58 }
59 return ContentDigest(doc.Snapshot, doc.DexKeys)
60}
61
62func OpenLocalRepo(stateDir string) (*git.Repository, error) {
63 path := filepath.Join(stateDir, RepoDir)
64 repo, err := git.PlainOpen(path)
65 if err == nil {
66 return repo, nil
67 }
68 if !errors.Is(err, git.ErrRepositoryNotExists) {
69 return nil, err
70 }
71 repo, err = git.PlainInit(path, true)
72 if err != nil {
73 return nil, err
74 }
75 if err := repo.Storer.SetReference(plumbing.NewSymbolicReference(plumbing.HEAD, replica.Branch)); err != nil {
76 return nil, err
77 }
78 return repo, nil
79}
80
81type checkpoints struct{ store *storage.Store }
82
83func (c checkpoints) ReadCheckpoint(ctx context.Context) (replica.Checkpoint, error) {
84 cp, err := c.store.UserbaseCommit(ctx)
85 return replica.Checkpoint{Value: cp.Value, Set: cp.Set}, err
86}
87
88func (c checkpoints) SwapCheckpoint(ctx context.Context, expected replica.Checkpoint, next string) error {
89 return c.store.CompareAndSwapUserbaseCommit(ctx, storage.Checkpoint{Value: expected.Value, Set: expected.Set}, next)
90}
91
92type ReplicatorConfig struct {
93 Store *storage.Store
94 StateDir string
95 Repo string
96 Mirror string
97 Keys Keys
98 ExportSigningKeys bool
99 Logger *slog.Logger
100 ObservePush func(at time.Time)
101}
102
103type Replicator struct {
104 store *storage.Store
105 stateDir string
106 repoDir string
107 primary string
108 mirror string
109 keys Keys
110 dex dexstorage.Storage
111 logger *slog.Logger
112 repo *git.Repository
113 sched *schedule
114 lastPush time.Time
115 pushed func(at time.Time)
116 inflight sync.WaitGroup
117
118 now func() time.Time
119 after func(time.Duration) <-chan time.Time
120 remoteHead func(ctx context.Context, url string, auth transport.AuthMethod, branch plumbing.ReferenceName) (plumbing.Hash, error)
121 planStart func(ctx context.Context, in replica.StartInput) (replica.StartPlan, error)
122 step func(ctx context.Context, in replica.StepInput) (replica.Result, error)
123 prune func(dir string, keep ...plumbing.Hash) (*git.Repository, error)
124}
125
126func NewReplicator(ctx context.Context, cfg ReplicatorConfig) (*Replicator, error) {
127 if cfg.Store == nil || cfg.StateDir == "" || cfg.Repo == "" || cfg.Keys.Identity == nil || cfg.Keys.BreakGlass == nil || cfg.Keys.Signer == nil {
128 return nil, errReplicatorConfig
129 }
130 repo, err := OpenLocalRepo(cfg.StateDir)
131 if err != nil {
132 return nil, err
133 }
134 var dex dexstorage.Storage
135 if cfg.ExportSigningKeys {
136 if dex, err = dexstore.Attach(ctx, cfg.Store); err != nil {
137 return nil, err
138 }
139 }
140 logger := cfg.Logger
141 if logger == nil {
142 logger = slog.New(slog.NewTextHandler(io.Discard, nil))
143 }
144 pushed := cfg.ObservePush
145 if pushed == nil {
146 pushed = func(time.Time) {}
147 }
148 if cp, err := cfg.Store.UserbaseCommit(ctx); err == nil && cp.Set {
149 if c, err := object.GetCommit(repo.Storer, plumbing.NewHash(cp.Value)); err == nil {
150 pushed(c.Committer.When)
151 }
152 }
153 return &Replicator{
154 store: cfg.Store,
155 stateDir: cfg.StateDir,
156 repoDir: filepath.Join(cfg.StateDir, RepoDir),
157 primary: cfg.Repo,
158 mirror: cfg.Mirror,
159 keys: cfg.Keys,
160 dex: dex,
161 logger: logger,
162 pushed: pushed,
163 repo: repo,
164 sched: newSchedule(func() float64 { return 2*rand.Float64() - 1 }),
165 now: time.Now,
166 after: time.After,
167 remoteHead: replica.RemoteHead,
168 planStart: replica.PlanStart,
169 step: replica.Step,
170 prune: replica.Prune,
171 }, nil
172}
173
174const maxActionsPerTick = 4
175
176var refusalKinds = []struct {
177 err error
178 kind string
179}{
180 {replica.ErrRewound, "rewound"},
181 {replica.ErrForeignChild, "foreign-child"},
182 {replica.ErrNotLocalHead, "not-local-head"},
183 {replica.ErrDiverged, "diverged"},
184 {replica.ErrRemoteEmpty, "remote-empty"},
185 {replica.ErrRemoteHasHistory, "remote-has-history"},
186 {replica.ErrOwnRootDiffers, "own-root-differs"},
187 {replica.ErrCheckpointMissing, "checkpoint-missing"},
188}
189
190func refusalKind(err error) string {
191 for _, k := range refusalKinds {
192 if errors.Is(err, k.err) {
193 return k.kind
194 }
195 }
196 return "refused"
197}
198
199func shortID(id plumbing.Hash) string {
200 if id.IsZero() {
201 return ""
202 }
203 return id.String()[:12]
204}
205
206func abandonable[T any](ctx context.Context, wg *sync.WaitGroup, f func(context.Context) (T, error)) (T, error) {
207 type outcome struct {
208 v T
209 err error
210 }
211 out := make(chan outcome, 1)
212 wg.Add(1)
213 go func() {
214 defer wg.Done()
215 v, err := f(ctx)
216 out <- outcome{v, err}
217 }()
218 select {
219 case o := <-out:
220 return o.v, o.err
221 case <-ctx.Done():
222 var zero T
223 return zero, ctx.Err()
224 }
225}
226
227type exportRead struct {
228 snapshot storage.UserbaseSnapshot
229 dexKeys json.RawMessage
230 digest [32]byte
231}
232
233func (r *Replicator) Run(ctx context.Context) {
234 for ctx.Err() == nil {
235 r.tick(ctx)
236 select {
237 case <-ctx.Done():
238 return
239 case <-r.after(pollInterval):
240 }
241 }
242}
243
244func (r *Replicator) tick(ctx context.Context) {
245 read, err := r.read(ctx)
246 if err != nil {
247 if ctx.Err() == nil {
248 r.logger.Warn("userbase export read failed", "err", err)
249 }
250 return
251 }
252 r.sched.observe(r.now(), read.digest)
253 for i := 0; i < maxActionsPerTick && ctx.Err() == nil; i++ {
254 now := r.now()
255 if r.sched.next(now) == actWait {
256 return
257 }
258 r.act(ctx, now, read)
259 }
260}
261
262func (r *Replicator) read(ctx context.Context) (exportRead, error) {
263 snapshot, err := r.store.ExportUserbase(ctx)
264 if err != nil {
265 return exportRead{}, err
266 }
267 var keys json.RawMessage
268 if r.dex != nil {
269 keys, err = dexstore.ExportPublicKeys(ctx, r.dex)
270 if errors.Is(err, dexstore.ErrNoKeys) {
271 keys, err = nil, nil
272 }
273 if err != nil {
274 return exportRead{}, err
275 }
276 }
277 digest, err := ContentDigest(snapshot, keys)
278 if err != nil {
279 return exportRead{}, err
280 }
281 return exportRead{snapshot: snapshot, dexKeys: keys, digest: digest}, nil
282}
283
284func (r *Replicator) act(ctx context.Context, now time.Time, read exportRead) {
285 primary, err := r.keys.Remote(r.primary)
286 if err != nil {
287 r.fail(now, err)
288 return
289 }
290 head, headErr := abandonable(ctx, &r.inflight, func(ctx context.Context) (plumbing.Hash, error) {
291 return r.remoteHead(ctx, primary.URL, primary.Auth, replica.Branch)
292 })
293 if ctx.Err() != nil {
294 return
295 }
296 if headErr != nil {
297 r.fail(now, headErr)
298 return
299 }
300 if refused, refusing := r.sched.currentRefusal(); refusing && head == refused.head {
301 r.sched.unchanged(now)
302 r.logger.Warn("userbase replication still refused", "kind", refused.kind, "remote", r.primary, "head", shortID(head), "last_push", r.lastPushText())
303 return
304 }
305 if err := replica.DropEmptyRefs(r.repoDir); err != nil {
306 r.fail(now, err)
307 return
308 }
309 if !r.sched.isPlanned {
310 r.plan(ctx, now, primary, head, read)
311 return
312 }
313 r.stepOnce(ctx, now, primary, head, read)
314}
315
316func (r *Replicator) plan(ctx context.Context, now time.Time, primary replica.Remote, head plumbing.Hash, read exportRead) {
317 plan, err := abandonable(ctx, &r.inflight, func(ctx context.Context) (replica.StartPlan, error) {
318 return r.planStart(ctx, replica.StartInput{
319 Repo: r.repo,
320 Dir: r.stateDir,
321 Checkpoints: checkpoints{r.store},
322 Primary: primary,
323 Digest: r.keys.Digest,
324 Current: read.digest,
325 })
326 })
327 if ctx.Err() != nil {
328 return
329 }
330 if err != nil {
331 r.settleError(ctx, now, head, err)
332 return
333 }
334 r.sched.planned(now, plan)
335 r.logger.Info("userbase replication planned", "remote", r.primary, "step_now", plan.StepNow, "pending", plan.Pending, "daily_due", plan.DailyDue, "checkpoint_unreadable", plan.CheckpointUnreadable)
336 if plan.CheckpointUnreadable {
337 r.logger.Warn("userbase checkpoint export cannot be read with the configured keys; committing a new export", "commit", r.checkpointID(ctx))
338 }
339}
340
341func (r *Replicator) checkpointID(ctx context.Context) string {
342 cp, err := r.store.UserbaseCommit(ctx)
343 if err != nil || !cp.Set {
344 return ""
345 }
346 return shortID(plumbing.NewHash(cp.Value))
347}
348
349func (r *Replicator) stepOnce(ctx context.Context, now time.Time, primary replica.Remote, head plumbing.Hash, read exportRead) {
350 plaintext, err := Encode(read.snapshot, now, read.dexKeys)
351 if err != nil {
352 r.fail(now, err)
353 return
354 }
355 blob, err := Seal(plaintext, r.keys.Identity.Recipient(), r.keys.BreakGlass)
356 if err != nil {
357 r.fail(now, err)
358 return
359 }
360 in := replica.StepInput{
361 Repo: r.repo,
362 Dir: r.stateDir,
363 Checkpoints: checkpoints{r.store},
364 Primary: primary,
365 Signer: r.keys.Signer,
366 Blob: blob,
367 When: now,
368 Digest: r.keys.Digest,
369 Current: read.digest,
370 }
371 var mirrorErr error
372 if r.mirror != "" {
373 mirror, err := r.keys.Remote(r.mirror)
374 if err != nil {
375 mirrorErr = err
376 } else {
377 in.Mirror = &mirror
378 }
379 }
380 res, err := abandonable(ctx, &r.inflight, func(ctx context.Context) (replica.Result, error) {
381 return r.step(ctx, in)
382 })
383 if ctx.Err() != nil {
384 return
385 }
386 if err != nil {
387 r.settleError(ctx, now, head, err)
388 return
389 }
390 if mirrorErr != nil && res.Outcome == replica.Pushed && res.MirrorErr == nil {
391 res.MirrorErr = mirrorErr
392 }
393 r.succeeded(now, res, read)
394}
395
396func (r *Replicator) settleError(ctx context.Context, now time.Time, head plumbing.Hash, err error) {
397 if !errors.Is(err, replica.ErrRefused) {
398 r.fail(now, err)
399 return
400 }
401 kind := refusalKind(err)
402 audit := r.sched.refused(now, refusal{kind: kind, head: head})
403 r.logger.Error("userbase replication refused", "kind", kind, "remote", r.primary, "head", shortID(head), "err", err, "retry_after", recheckInterval, "last_push", r.lastPushText())
404 if !audit {
405 return
406 }
407 detail, err := json.Marshal(map[string]string{"kind": kind, "remote": r.primary, "head": shortID(head)})
408 if err == nil {
409 _, err = r.store.AppendAudit(ctx, storage.AppendAuditInput{Actor: "system", Action: "userbase.refused", Subject: "userbase", Detail: string(detail)})
410 }
411 if err != nil {
412 r.logger.Error("userbase refusal audit failed", "kind", kind, "err", err)
413 }
414}
415
416func (r *Replicator) fail(now time.Time, err error) {
417 r.sched.failed(now, err)
418 r.logger.Warn("userbase replication attempt failed", "remote", r.primary, "err", err, "retry_after", r.sched.retryAt().Sub(now).Round(time.Second), "last_push", r.lastPushText())
419}
420
421func (r *Replicator) succeeded(now time.Time, res replica.Result, read exportRead) {
422 committed := read.digest
423 if res.Outcome != replica.Pushed {
424 committed = r.committedDigest(res.Commit)
425 }
426 commitTime := now
427 if c, err := object.GetCommit(r.repo.Storer, res.Commit); err == nil {
428 commitTime = c.Committer.When
429 }
430 r.sched.succeeded(now, read.digest, committed, commitTime)
431 if res.Outcome != replica.Pushed {
432 r.logger.Info("userbase push", "remote", r.primary, "role", "repo", "result", "ok", "outcome", "adopted", "commit", res.Commit.String(), "last_push", r.lastPushText())
433 return
434 }
435 r.lastPush = now
436 r.pushed(now)
437 r.logger.Info("userbase export committed", "commit", res.Commit.String(), "remote", r.primary)
438 r.logger.Info("userbase push", "remote", r.primary, "role", "repo", "result", "ok", "outcome", "pushed", "commit", res.Commit.String(), "last_push", r.lastPushText())
439 if r.mirror != "" {
440 if res.MirrorErr != nil {
441 r.logger.Warn("userbase push", "remote", r.mirror, "role", "mirror", "result", "failed", "reseeded", res.MirrorReseeded, "err", res.MirrorErr, "last_push", r.lastPushText())
442 } else {
443 r.logger.Info("userbase push", "remote", r.mirror, "role", "mirror", "result", "ok", "reseeded", res.MirrorReseeded, "commit", res.Commit.String(), "last_push", r.lastPushText())
444 }
445 }
446 if r.sched.pruneDue(now, res) {
447 r.pruneLocal(now, res.Commit)
448 }
449}
450
451func (r *Replicator) committedDigest(id plumbing.Hash) [32]byte {
452 blob, err := replica.ReadFile(r.repo.Storer, id)
453 if err != nil {
454 return [32]byte{}
455 }
456 digest, err := r.keys.Digest(blob)
457 if err != nil {
458 return [32]byte{}
459 }
460 return digest
461}
462
463func (r *Replicator) pruneLocal(now time.Time, keep plumbing.Hash) {
464 r.sched.pruned(now)
465 repo, err := r.prune(r.repoDir, keep)
466 if err == nil {
467 r.repo = repo
468 r.logger.Info("userbase local repository pruned", "kept", keep.String())
469 return
470 }
471 r.logger.Warn("userbase local repository prune failed", "err", err)
472 if repo, err = git.PlainOpen(r.repoDir); err != nil {
473 r.logger.Error("userbase local repository reopen failed", "err", err)
474 return
475 }
476 r.repo = repo
477}
478
479func (r *Replicator) lastPushText() string {
480 if r.lastPush.IsZero() {
481 return "never"
482 }
483 return r.lastPush.UTC().Format(time.RFC3339)
484}