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// deleteChunk is how many merge request keys one batch deletes before the
94// last one, which also removes the repository.
95const deleteChunk = 1000
96
97// deleteTries bounds the listings of Delete: a listing is repeated when a merge
98// request write got in after it.
99const deleteTries = 32
100
101// Delete removes the repository and everything keyed to it: its merge
102// requests, their comments and read positions, and its number sequence. The
103// Git data is moved to the trash by the caller; the merge requests are not
104// kept anywhere, like the repository metadata, so a repository created again
105// under the same name starts clean, with its numbers starting at 1. A missing
106// repository is not an error.
107//
108// The merge request keys go first, in chunks, and the repository key goes with
109// the last chunk: a delete that fails half way leaves the repository, which
110// the caller can delete again. The last batch requires the repository key to
111// be at the version the keys were listed at. Every write of a merge request key
112// rewrites the repository key (see MergeRequests.guardRepo), so a write that
113// got in after the listing makes the batch fail and the keys are listed again,
114// and a write after the batch finds no repository: no key outlives the
115// repository to attach itself to a repository created again.
116//
117//hexcheck:ignore suspicious-business-logic-in-adapter chunked delete under a version precondition: a storage concern, no policy
118func (s *Store) Delete(ctx context.Context, name domain.RepoName) error {
119 for range deleteTries {
120 snap := s.db.Snapshot(ctx)
121 repoItem, repoOK := snap.Get(ctx, repoKey(name))
122 keys := mergeRequestKeys(ctx, snap, name)
123 for len(keys) > deleteChunk {
124 var b kv.Batch
125 for _, k := range keys[:deleteChunk] {
126 b.Delete(k)
127 }
128 if _, err := s.db.Apply(ctx, &b); err != nil {
129 return fmt.Errorf("delete merge requests of %q: %w", name, err)
130 }
131 keys = keys[deleteChunk:]
132 }
133 var b kv.Batch
134 if repoOK {
135 b.RequireVersion(repoKey(name), repoItem.Version)
136 } else {
137 b.RequireAbsent(repoKey(name))
138 }
139 for _, k := range keys {
140 b.Delete(k)
141 }
142 b.Delete(repoKey(name))
143 _, err := s.db.Apply(ctx, &b)
144 if errors.Is(err, kv.ErrConflict) {
145 continue
146 }
147 if err != nil {
148 return fmt.Errorf("delete repository %q: %w", name, err)
149 }
150 return nil
151 }
152 return fmt.Errorf("delete repository %q: %w", name, domain.ErrConflict)
153}
154
155func encodeRepo(r domain.Repo) ([]byte, error) {
156 if r.Name.IsZero() {
157 return nil, fmt.Errorf("%w: empty name", domain.ErrInvalidName)
158 }
159 if !utf8.ValidString(r.Description) {
160 return nil, errors.New("repository description is not valid UTF-8")
161 }
162 vis, err := encodeVisibility(r.Visibility)
163 if err != nil {
164 return nil, err
165 }
166 settings := r.Settings
167 if settings.IsZero() {
168 // Settings are always stored explicitly, so that a later change of
169 // the defaults never alters a stored repository.
170 settings = domain.DefaultRepoSettings()
171 }
172 pbSettings, err := encodeSettings(settings)
173 if err != nil {
174 return nil, err
175 }
176 pb := &storagepb.Repo{
177 Name: r.Name.String(),
178 Description: r.Description,
179 Visibility: vis,
180 CreatedAt: encodeTime(r.CreatedAt),
181 PushedAt: encodeTime(r.PushedAt),
182 Settings: pbSettings,
183 }
184 return mustMarshal(pb), nil
185}
186
187func decodeRepo(val []byte) (domain.Repo, error) {
188 var pb storagepb.Repo
189 if err := proto.Unmarshal(val, &pb); err != nil {
190 return domain.Repo{}, fmt.Errorf("decode repository: %w", err)
191 }
192 name, err := domain.ParseRepoName(pb.GetName())
193 if err != nil {
194 return domain.Repo{}, fmt.Errorf("decode repository: %w", err)
195 }
196 return domain.Repo{
197 Name: name,
198 Description: pb.GetDescription(),
199 Visibility: decodeVisibility(pb.GetVisibility()),
200 CreatedAt: decodeTime(pb.GetCreatedAt()),
201 PushedAt: decodeTime(pb.GetPushedAt()),
202 Settings: decodeSettings(pb.GetSettings()),
203 }, nil
204}
205
206func encodeVisibility(v domain.Visibility) (storagepb.Visibility, error) {
207 switch v {
208 case domain.VisibilityPublic:
209 return storagepb.Visibility_VISIBILITY_PUBLIC, nil
210 case domain.VisibilityPrivate:
211 return storagepb.Visibility_VISIBILITY_PRIVATE, nil
212 }
213 return storagepb.Visibility_VISIBILITY_UNSPECIFIED, fmt.Errorf("%w: %q", domain.ErrInvalidVisibility, v)
214}
215
216// decodeVisibility maps anything but public to private, including the
217// unspecified value and values from a newer schema: the safe side.
218func decodeVisibility(v storagepb.Visibility) domain.Visibility {
219 if v == storagepb.Visibility_VISIBILITY_PUBLIC {
220 return domain.VisibilityPublic
221 }
222 return domain.VisibilityPrivate
223}
224
225func encodeTime(t time.Time) *timestamppb.Timestamp {
226 if t.IsZero() {
227 return nil
228 }
229 return timestamppb.New(t)
230}
231
232func decodeTime(ts *timestamppb.Timestamp) time.Time {
233 if ts == nil {
234 return time.Time{}
235 }
236 return ts.AsTime()
237}