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 and run numbers. 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 for len(keys) > deleteChunk {
148 var b kv.Batch
149 for _, k := range keys[:deleteChunk] {
150 b.Delete(k)
151 }
152 if _, err := s.db.Apply(ctx, &b); err != nil {
153 return fmt.Errorf("delete merge requests of %q: %w", name, err)
154 }
155 keys = keys[deleteChunk:]
156 }
157 var b kv.Batch
158 if repoOK {
159 b.RequireVersion(repoKey(name), repoItem.Version)
160 } else {
161 b.RequireAbsent(repoKey(name))
162 }
163 for _, k := range keys {
164 b.Delete(k)
165 }
166 b.Delete(repoKey(name))
167 _, err := s.db.Apply(ctx, &b)
168 if errors.Is(err, kv.ErrConflict) {
169 continue
170 }
171 if err != nil {
172 return fmt.Errorf("delete repository %q: %w", name, err)
173 }
174 return nil
175 }
176 return fmt.Errorf("delete repository %q: %w", name, domain.ErrConflict)
177}
178
179func encodeRepo(r domain.Repo) ([]byte, error) {
180 if r.Name.IsZero() {
181 return nil, fmt.Errorf("%w: empty name", domain.ErrInvalidName)
182 }
183 if !utf8.ValidString(r.Description) {
184 return nil, errors.New("repository description is not valid UTF-8")
185 }
186 vis, err := encodeVisibility(r.Visibility)
187 if err != nil {
188 return nil, err
189 }
190 settings := r.Settings
191 if settings.IsZero() {
192 // Settings are always stored explicitly, so that a later change of
193 // the defaults never alters a stored repository.
194 settings = domain.DefaultRepoSettings()
195 }
196 pbSettings, err := encodeSettings(settings)
197 if err != nil {
198 return nil, err
199 }
200 pb := &storagepb.Repo{
201 Name: r.Name.String(),
202 Description: r.Description,
203 Visibility: vis,
204 CreatedAt: encodeTime(r.CreatedAt),
205 PushedAt: encodeTime(r.PushedAt),
206 Settings: pbSettings,
207 }
208 return mustMarshal(pb), nil
209}
210
211func decodeRepo(val []byte) (domain.Repo, error) {
212 var pb storagepb.Repo
213 if err := proto.Unmarshal(val, &pb); err != nil {
214 return domain.Repo{}, fmt.Errorf("decode repository: %w", err)
215 }
216 name, err := domain.ParseRepoName(pb.GetName())
217 if err != nil {
218 return domain.Repo{}, fmt.Errorf("decode repository: %w", err)
219 }
220 return domain.Repo{
221 Name: name,
222 Description: pb.GetDescription(),
223 Visibility: decodeVisibility(pb.GetVisibility()),
224 CreatedAt: decodeTime(pb.GetCreatedAt()),
225 PushedAt: decodeTime(pb.GetPushedAt()),
226 Settings: decodeSettings(pb.GetSettings()),
227 }, nil
228}
229
230func encodeVisibility(v domain.Visibility) (storagepb.Visibility, error) {
231 switch v {
232 case domain.VisibilityPublic:
233 return storagepb.Visibility_VISIBILITY_PUBLIC, nil
234 case domain.VisibilityPrivate:
235 return storagepb.Visibility_VISIBILITY_PRIVATE, nil
236 }
237 return storagepb.Visibility_VISIBILITY_UNSPECIFIED, fmt.Errorf("%w: %q", domain.ErrInvalidVisibility, v)
238}
239
240// decodeVisibility maps anything but public to private, including the
241// unspecified value and values from a newer schema: the safe side.
242func decodeVisibility(v storagepb.Visibility) domain.Visibility {
243 if v == storagepb.Visibility_VISIBILITY_PUBLIC {
244 return domain.VisibilityPublic
245 }
246 return domain.VisibilityPrivate
247}
248
249func encodeTime(t time.Time) *timestamppb.Timestamp {
250 if t.IsZero() {
251 return nil
252 }
253 return timestamppb.New(t)
254}
255
256func decodeTime(ts *timestamppb.Timestamp) time.Time {
257 if ts == nil {
258 return time.Time{}
259 }
260 return ts.AsTime()
261}