1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package kvstore
4
5import (
6 "context"
7 "errors"
8 "fmt"
9
10 kv "bnema.dev/ermite-kv"
11 "google.golang.org/protobuf/proto"
12
13 storagepb "bnema.dev/ermite/gen/ermite/storage"
14 "bnema.dev/ermite/internal/domain"
15 "bnema.dev/ermite/internal/ports"
16)
17
18// Who moves a branch or a tag to its object (CI-3, D1; domain.CIRefTip):
19//
20// cireftip/<repo>/<full ref name> CIRefTip
21//
22// Written under the guard of the repository key, deleted with the repository
23// (ciKeys). Ref names hold '/', so the key is split at the first '/' after the
24// repository, whose name holds none.
25const refTipPrefix = "cireftip/"
26
27var _ ports.CIRefTipStore = (*CI)(nil)
28
29func refTipKey(repo domain.RepoName, ref string) []byte {
30 return []byte(refTipPrefix + repo.String() + "/" + ref)
31}
32
33func refTipsPrefix(repo domain.RepoName) string { return refTipPrefix + repo.String() + "/" }
34
35// updateAttemptsTip bounds the retries of a write that conflicts with another
36// write of the repository.
37const updateAttemptsTip = 8
38
39// PutRefTip records who moves the ref to its object. A repository that does not
40// exist is domain.ErrNotFound.
41func (c *CI) PutRefTip(ctx context.Context, t domain.CIRefTip) error {
42 val, err := proto.Marshal(&storagepb.CIRefTip{
43 Sha: t.Sha, Actor: t.Actor, Trusted: t.Trusted,
44 PusherKind: string(t.Pusher.Kind), PusherTokenId: t.Pusher.TokenID,
45 })
46 if err != nil {
47 return fmt.Errorf("encode ref tip: %w", err)
48 }
49 for range updateAttemptsTip {
50 var b kv.Batch
51 if err := c.guardRepo(ctx, &b, t.Repo); err != nil {
52 return err
53 }
54 b.Put(refTipKey(t.Repo, t.Ref), val)
55 _, err := c.db.Apply(ctx, &b)
56 if errors.Is(err, kv.ErrConflict) {
57 continue
58 }
59 if err != nil {
60 return fmt.Errorf("store ref tip: %w", err)
61 }
62 return nil
63 }
64 return fmt.Errorf("store ref tip: %w", domain.ErrConflict)
65}
66
67// GetRefTip reads the record of a ref, or domain.ErrNotFound.
68func (c *CI) GetRefTip(ctx context.Context, repo domain.RepoName, ref string) (domain.CIRefTip, error) {
69 it, ok := c.db.Get(ctx, refTipKey(repo, ref))
70 if !ok {
71 return domain.CIRefTip{}, domain.ErrNotFound
72 }
73 var pb storagepb.CIRefTip
74 if err := proto.Unmarshal(it.Value, &pb); err != nil {
75 return domain.CIRefTip{}, fmt.Errorf("decode ref tip: %w", err)
76 }
77 return domain.CIRefTip{
78 Repo: repo, Ref: ref, Sha: pb.GetSha(), Actor: pb.GetActor(), Trusted: pb.GetTrusted(),
79 Pusher: domain.CIPusher{Kind: domain.CIPusherKind(pb.GetPusherKind()), TokenID: pb.GetPusherTokenId()},
80 }, nil
81}
82
83// DeleteRefTip removes the record of a ref (the ref is deleted).
84func (c *CI) DeleteRefTip(ctx context.Context, repo domain.RepoName, ref string) error {
85 var b kv.Batch
86 b.Delete(refTipKey(repo, ref))
87 if _, err := c.db.Apply(ctx, &b); err != nil {
88 return fmt.Errorf("delete ref tip: %w", err)
89 }
90 return nil
91}
92
93// DeleteRefTipIf removes the record of a ref only when it still names sha.
94func (c *CI) DeleteRefTipIf(ctx context.Context, repo domain.RepoName, ref, sha string) error {
95 for range updateAttemptsTip {
96 it, ok := c.db.Get(ctx, refTipKey(repo, ref))
97 if !ok {
98 return nil
99 }
100 var pb storagepb.CIRefTip
101 if err := proto.Unmarshal(it.Value, &pb); err != nil {
102 return fmt.Errorf("decode ref tip: %w", err)
103 }
104 if pb.GetSha() != sha {
105 return nil // a newer push owns the record
106 }
107 var b kv.Batch
108 b.RequireVersion(refTipKey(repo, ref), it.Version)
109 b.Delete(refTipKey(repo, ref))
110 _, err := c.db.Apply(ctx, &b)
111 if errors.Is(err, kv.ErrConflict) {
112 continue
113 }
114 if err != nil {
115 return fmt.Errorf("delete ref tip: %w", err)
116 }
117 return nil
118 }
119 return fmt.Errorf("delete ref tip: %w", domain.ErrConflict)
120}