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