1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package kvstore
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "slices"
10 "time"
11 "unicode/utf8"
12
13 kv "bnema.dev/ermite-kv"
14 "google.golang.org/protobuf/proto"
15 "google.golang.org/protobuf/types/known/timestamppb"
16
17 storagepb "bnema.dev/ermite/gen/ermite/storage"
18 "bnema.dev/ermite/internal/domain"
19)
20
21const repoPrefix = "repo/"
22
23func repoKey(name domain.RepoName) []byte { return []byte(repoPrefix + name.String()) }
24
25// Create stores repo with an "absent" precondition: a repository with that
26// name, or a concurrent Create that commits first, yields domain.ErrAlreadyExists.
27func (s *Store) Create(ctx context.Context, repo domain.Repo) error {
28 val, err := encodeRepo(repo)
29 if err != nil {
30 return err
31 }
32 key := repoKey(repo.Name)
33 var b kv.Batch
34 b.RequireAbsent(key)
35 b.Put(key, val)
36 if _, err := s.db.Apply(ctx, &b); err != nil {
37 if errors.Is(err, kv.ErrConflict) {
38 return domain.ErrAlreadyExists
39 }
40 return fmt.Errorf("create repository %q: %w", repo.Name, err)
41 }
42 return nil
43}
44
45// Get returns the repository or domain.ErrNotFound.
46func (s *Store) Get(ctx context.Context, name domain.RepoName) (domain.Repo, error) {
47 item, ok := s.db.Get(ctx, repoKey(name))
48 if !ok {
49 return domain.Repo{}, domain.ErrNotFound
50 }
51 return decodeRepo(item.Value)
52}
53
54// List returns every repository ordered by name.
55func (s *Store) List(ctx context.Context) ([]domain.Repo, error) {
56 var repos []domain.Repo
57 for key, item := range s.db.Snapshot(ctx).Prefix(ctx, []byte(repoPrefix)) {
58 repo, err := decodeRepo(item.Value)
59 if err != nil {
60 return nil, fmt.Errorf("decode %s: %w", key, err)
61 }
62 repos = append(repos, repo)
63 }
64 return repos, nil
65}
66
67const repoMutationTries = 8
68
69// Mutate applies a retry-safe mutation under the version it read.
70//
71//hexcheck:ignore suspicious-business-logic-in-adapter optimistic concurrency is a storage concern
72func (s *Store) Mutate(ctx context.Context, name domain.RepoName, fn func(*domain.Repo) (bool, error)) (domain.Repo, error) {
73 key := repoKey(name)
74 for range repoMutationTries {
75 if err := ctx.Err(); err != nil {
76 return domain.Repo{}, err
77 }
78 item, ok := s.db.Get(ctx, key)
79 if !ok {
80 return domain.Repo{}, domain.ErrNotFound
81 }
82 repo, err := decodeRepo(item.Value)
83 if err != nil {
84 return domain.Repo{}, err
85 }
86 changed, err := fn(&repo)
87 if err != nil {
88 return domain.Repo{}, err
89 }
90 if repo.Name != name {
91 return domain.Repo{}, fmt.Errorf("mutate repository name: %w", domain.ErrInvalidName)
92 }
93 if !changed {
94 return decodeRepo(item.Value)
95 }
96 val, err := encodeRepo(repo)
97 if err != nil {
98 return domain.Repo{}, err
99 }
100 var b kv.Batch
101 b.RequireVersion(key, item.Version)
102 b.Put(key, val)
103 if _, err := s.db.Apply(ctx, &b); err != nil {
104 if errors.Is(err, kv.ErrConflict) {
105 continue
106 }
107 return domain.Repo{}, fmt.Errorf("mutate repository %q: %w", name, err)
108 }
109 return decodeRepo(val)
110 }
111 return domain.Repo{}, domain.ErrConflict
112}
113
114// guardRepoKey makes the batch depend on the repository: it requires the
115// repository key to be at the version just read and rewrites it, which gives it
116// a new version. Every write that adds a key under the repository (a merge
117// request, a comment, a read position, a CI run, job or check) goes through it,
118// so that
119//
120// - such a write fails with domain.ErrNotFound once the repository is deleted,
121// and leaves no key that a repository created again under the name would
122// inherit;
123// - Store.Delete, which requires the version of the repository key it listed
124// the keys at, notices a write that got in after the listing and lists again.
125//
126// It returns domain.ErrNotFound when the repository does not exist.
127func guardRepoKey(ctx context.Context, db *kv.DB, b *kv.Batch, repo domain.RepoName) error {
128 item, ok := db.Get(ctx, repoKey(repo))
129 if !ok {
130 return domain.ErrNotFound
131 }
132 b.RequireVersion(repoKey(repo), item.Version)
133 b.Put(repoKey(repo), item.Value)
134 return nil
135}
136
137// deleteChunk is how many merge request keys one batch deletes before the
138// last one, which also removes the repository.
139const deleteChunk = 1000
140
141// deleteTries bounds the listings of Delete: a listing is repeated when a merge
142// request write got in after it.
143const deleteTries = 32
144
145// Delete removes the repository and everything keyed to it: its merge
146// requests, their comments and read positions, and its number sequence, and
147// its CI runs, jobs, checks, run numbers and artifacts. The
148// Git data is moved to the trash by the caller; the merge requests are not
149// kept anywhere, like the repository metadata, so a repository created again
150// under the same name starts clean, with its numbers starting at 1. A missing
151// repository is not an error.
152//
153// The merge request keys go first, in chunks, and the repository key goes with
154// the last chunk: a delete that fails half way leaves the repository, which
155// the caller can delete again. The last batch requires the repository key to
156// be at the version the keys were listed at. Every write of a merge request key
157// rewrites the repository key (see guardRepoKey), so a write that
158// got in after the listing makes the batch fail and the keys are listed again,
159// and a write after the batch finds no repository: no key outlives the
160// repository to attach itself to a repository created again.
161//
162//hexcheck:ignore suspicious-business-logic-in-adapter chunked delete under a version precondition: a storage concern, no policy
163func (s *Store) Delete(ctx context.Context, name domain.RepoName) error {
164 for range deleteTries {
165 snap := s.db.Snapshot(ctx)
166 repoItem, repoOK := snap.Get(ctx, repoKey(name))
167 keys := append(mergeRequestKeys(ctx, snap, name), issueKeys(ctx, snap, name)...)
168 keys = append(keys, pendingKeys(ctx, snap, name)...)
169 keys = append(keys, ciKeys(ctx, snap, name)...)
170 keys = append(keys, settingsKeys(ctx, snap, name)...)
171 for len(keys) > deleteChunk {
172 var b kv.Batch
173 for _, k := range keys[:deleteChunk] {
174 b.Delete(k)
175 }
176 if _, err := s.db.Apply(ctx, &b); err != nil {
177 return fmt.Errorf("delete merge requests of %q: %w", name, err)
178 }
179 keys = keys[deleteChunk:]
180 }
181 var b kv.Batch
182 if repoOK {
183 b.RequireVersion(repoKey(name), repoItem.Version)
184 } else {
185 b.RequireAbsent(repoKey(name))
186 }
187 for _, k := range keys {
188 b.Delete(k)
189 }
190 b.Delete(repoKey(name))
191 _, err := s.db.Apply(ctx, &b)
192 if errors.Is(err, kv.ErrConflict) {
193 continue
194 }
195 if err != nil {
196 return fmt.Errorf("delete repository %q: %w", name, err)
197 }
198 return nil
199 }
200 return fmt.Errorf("delete repository %q: %w", name, domain.ErrConflict)
201}
202
203func encodeRepo(r domain.Repo) ([]byte, error) {
204 if r.Name.IsZero() {
205 return nil, fmt.Errorf("%w: empty name", domain.ErrInvalidName)
206 }
207 if !utf8.ValidString(r.Description) {
208 return nil, errors.New("repository description is not valid UTF-8")
209 }
210 vis, err := encodeVisibility(r.Visibility)
211 if err != nil {
212 return nil, err
213 }
214 settings := r.Settings
215 if settings.IsZero() {
216 // Settings are always stored explicitly, so that a later change of
217 // the defaults never alters a stored repository.
218 settings = domain.DefaultRepoSettings()
219 }
220 pbSettings, err := encodeSettings(settings)
221 if err != nil {
222 return nil, err
223 }
224 pb := &storagepb.Repo{
225 Name: r.Name.String(),
226 Description: r.Description,
227 Topics: r.Topics,
228 Visibility: vis,
229 CreatedAt: encodeTime(r.CreatedAt),
230 PushedAt: encodeTime(r.PushedAt),
231 ActiveAt: encodeTime(r.ActiveAt),
232 Settings: pbSettings,
233 }
234 return mustMarshal(pb), nil
235}
236
237func decodeRepo(val []byte) (domain.Repo, error) {
238 var pb storagepb.Repo
239 if err := proto.Unmarshal(val, &pb); err != nil {
240 return domain.Repo{}, fmt.Errorf("decode repository: %w", err)
241 }
242 name, err := domain.ParseRepoName(pb.GetName())
243 if err != nil {
244 return domain.Repo{}, fmt.Errorf("decode repository: %w", err)
245 }
246 return domain.Repo{
247 Name: name,
248 Description: pb.GetDescription(),
249 Topics: slices.Clone(pb.GetTopics()),
250 Visibility: decodeVisibility(pb.GetVisibility()),
251 CreatedAt: decodeTime(pb.GetCreatedAt()),
252 PushedAt: decodeTime(pb.GetPushedAt()),
253 ActiveAt: decodeTime(pb.GetActiveAt()),
254 Settings: decodeSettings(pb.GetSettings()),
255 }, nil
256}
257
258func encodeVisibility(v domain.Visibility) (storagepb.Visibility, error) {
259 switch v {
260 case domain.VisibilityPublic:
261 return storagepb.Visibility_VISIBILITY_PUBLIC, nil
262 case domain.VisibilityPrivate:
263 return storagepb.Visibility_VISIBILITY_PRIVATE, nil
264 }
265 return storagepb.Visibility_VISIBILITY_UNSPECIFIED, fmt.Errorf("%w: %q", domain.ErrInvalidVisibility, v)
266}
267
268// decodeVisibility maps anything but public to private, including the
269// unspecified value and values from a newer schema: the safe side.
270func decodeVisibility(v storagepb.Visibility) domain.Visibility {
271 if v == storagepb.Visibility_VISIBILITY_PUBLIC {
272 return domain.VisibilityPublic
273 }
274 return domain.VisibilityPrivate
275}
276
277func encodeTime(t time.Time) *timestamppb.Timestamp {
278 if t.IsZero() {
279 return nil
280 }
281 return timestamppb.New(t)
282}
283
284func decodeTime(ts *timestamppb.Timestamp) time.Time {
285 if ts == nil {
286 return time.Time{}
287 }
288 return ts.AsTime()
289}