1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package kvstore_test
4
5import (
6 "context"
7 "errors"
8 "strings"
9 "sync"
10 "testing"
11 "time"
12
13 "github.com/stretchr/testify/assert"
14 "github.com/stretchr/testify/require"
15 "google.golang.org/protobuf/proto"
16
17 kv "bnema.dev/ermite-kv"
18
19 storagepb "bnema.dev/ermite/gen/ermite/storage"
20 "bnema.dev/ermite/internal/adapters/kvstore"
21 "bnema.dev/ermite/internal/domain"
22)
23
24func newMR(t *testing.T, repo domain.RepoName, title string) domain.MergeRequest {
25 t.Helper()
26 ti, err := domain.ParseMRTitle(title)
27 require.NoError(t, err)
28 desc, err := domain.ParsePlainText("why\nand how")
29 require.NoError(t, err)
30 return domain.MergeRequest{
31 Repo: repo, Title: ti, Description: desc, Source: "feat/x", Target: "main",
32 State: domain.MRStateOpen, Author: "agent", CreatedAt: created, UpdatedAt: created,
33 }
34}
35
36func oid(c byte) domain.ObjectID {
37 id, err := domain.ParseObjectID(strings.Repeat(string(c), 40))
38 if err != nil {
39 panic(err)
40 }
41 return id
42}
43
44// repoWithMRs opens a store holding repository "proj".
45func repoWithMRs(t *testing.T) (*kvstore.Store, *kvstore.MergeRequests, domain.RepoName) {
46 t.Helper()
47 s := openStore(t, t.TempDir())
48 n := name(t, "proj")
49 require.NoError(t, s.Create(context.Background(), domain.Repo{Name: n, Visibility: domain.VisibilityPrivate, CreatedAt: created}))
50 return s, s.MergeRequests(), n
51}
52
53func TestMergeRequestNumbersAreSequentialPerRepository(t *testing.T) {
54 ctx := context.Background()
55 s, mrs, n := repoWithMRs(t)
56 other := name(t, "other")
57 require.NoError(t, s.Create(ctx, domain.Repo{Name: other, CreatedAt: created, Visibility: domain.VisibilityPublic}))
58
59 for want := int64(1); want <= 3; want++ {
60 got, err := mrs.Create(ctx, newMR(t, n, "t"))
61 require.NoError(t, err)
62 assert.Equal(t, want, got.Number)
63 }
64 got, err := mrs.Create(ctx, newMR(t, other, "t"))
65 require.NoError(t, err)
66 assert.Equal(t, int64(1), got.Number, "each repository counts on its own")
67
68 _, err = mrs.Create(ctx, newMR(t, name(t, "ghost"), "t"))
69 assert.ErrorIs(t, err, domain.ErrNotFound, "no merge request for a missing repository")
70}
71
72func TestMergeRequestRoundTrip(t *testing.T) {
73 ctx := context.Background()
74 _, mrs, n := repoWithMRs(t)
75 in := newMR(t, n, "Ajoute « ça »")
76 in.HeadID, in.BaseID, in.MergedCommitID = oid('a'), oid('b'), oid('c')
77 in.MergeMode = domain.MergeModeSquash
78 in.State = domain.MRStateMerged
79 in.CommentSeq = 4
80 created1, err := mrs.Create(ctx, in)
81 require.NoError(t, err)
82
83 got, err := mrs.Get(ctx, n, created1.Number)
84 require.NoError(t, err)
85 assert.NotZero(t, got.Version)
86 in.Number, in.Version = 1, got.Version
87 assert.Equal(t, in, got)
88
89 _, err = mrs.Get(ctx, n, 2)
90 assert.ErrorIs(t, err, domain.ErrNotFound)
91 _, err = mrs.Get(ctx, n, 0)
92 assert.ErrorIs(t, err, domain.ErrNotFound)
93 _, err = mrs.Get(ctx, n, -3)
94 assert.ErrorIs(t, err, domain.ErrNotFound)
95}
96
97func TestMergeRequestListIsNewestFirstAndPerRepository(t *testing.T) {
98 ctx := context.Background()
99 s, mrs, n := repoWithMRs(t)
100 // "proj-2" shares the prefix "proj": its keys must not leak into proj's.
101 sib := name(t, "proj-2")
102 require.NoError(t, s.Create(ctx, domain.Repo{Name: sib, CreatedAt: created, Visibility: domain.VisibilityPublic}))
103 for range 12 { // beyond 10: numbers sort numerically, not as text
104 _, err := mrs.Create(ctx, newMR(t, n, "t"))
105 require.NoError(t, err)
106 }
107 _, err := mrs.Create(ctx, newMR(t, sib, "sibling"))
108 require.NoError(t, err)
109
110 list, err := mrs.List(ctx, n)
111 require.NoError(t, err)
112 require.Len(t, list, 12)
113 for i, mr := range list {
114 assert.Equal(t, int64(12-i), mr.Number)
115 assert.Equal(t, n, mr.Repo)
116 }
117 list, err = mrs.List(ctx, name(t, "none"))
118 require.NoError(t, err)
119 assert.Empty(t, list)
120}
121
122func TestConcurrentCreateNeverSharesANumber(t *testing.T) {
123 ctx := context.Background()
124 _, mrs, n := repoWithMRs(t)
125 const workers = 8
126 var (
127 wg sync.WaitGroup
128 mu sync.Mutex
129 numbers []int64
130 )
131 for range workers {
132 wg.Go(func() {
133 got, err := mrs.Create(ctx, newMR(t, n, "t"))
134 if assert.NoError(t, err) {
135 mu.Lock()
136 numbers = append(numbers, got.Number)
137 mu.Unlock()
138 }
139 })
140 }
141 wg.Wait()
142 assert.ElementsMatch(t, []int64{1, 2, 3, 4, 5, 6, 7, 8}, numbers)
143}
144
145func TestMergeRequestUpdateNeedsTheVersionItWasReadAt(t *testing.T) {
146 ctx := context.Background()
147 _, mrs, n := repoWithMRs(t)
148 first, err := mrs.Create(ctx, newMR(t, n, "t"))
149 require.NoError(t, err)
150
151 a, err := mrs.Get(ctx, n, first.Number)
152 require.NoError(t, err)
153 b := a
154 a.State = domain.MRStateClosed
155 require.NoError(t, mrs.Update(ctx, a))
156
157 b.State = domain.MRStateMerged
158 assert.ErrorIs(t, mrs.Update(ctx, b), domain.ErrConflict, "b was read before a wrote")
159
160 got, err := mrs.Get(ctx, n, first.Number)
161 require.NoError(t, err)
162 assert.Equal(t, domain.MRStateClosed, got.State)
163
164 gone := got
165 gone.Number = 99
166 assert.ErrorIs(t, mrs.Update(ctx, gone), domain.ErrNotFound)
167 fresh := newMR(t, n, "never read")
168 fresh.Number = first.Number
169 assert.ErrorIs(t, mrs.Update(ctx, fresh), domain.ErrConflict, "an unread request has no version")
170}
171
172func newComment(t *testing.T, id int64, text string) domain.Comment {
173 t.Helper()
174 pt, err := domain.ParsePlainText(text)
175 require.NoError(t, err)
176 return domain.Comment{
177 ID: id, Author: "root", Path: "a.txt", Line: 3, Side: domain.SideNew,
178 CommitID: oid('d'), Text: pt, CreatedAt: created.Add(time.Duration(id) * time.Minute),
179 }
180}
181
182func TestComments(t *testing.T) {
183 ctx := context.Background()
184 _, mrs, n := repoWithMRs(t)
185 first, err := mrs.Create(ctx, newMR(t, n, "t"))
186 require.NoError(t, err)
187
188 // Twelve comments: ids sort numerically too.
189 for id := int64(1); id <= 12; id++ {
190 mr, err := mrs.Get(ctx, n, first.Number)
191 require.NoError(t, err)
192 mr.CommentSeq = id
193 c := newComment(t, id, "note")
194 if id == 2 {
195 c.Path, c.Line, c.Side = "", 0, domain.SideNone
196 }
197 require.NoError(t, mrs.SaveComment(ctx, mr, c))
198 }
199 list, err := mrs.ListComments(ctx, n, first.Number)
200 require.NoError(t, err)
201 require.Len(t, list, 12)
202 for i, c := range list {
203 assert.Equal(t, int64(i+1), c.ID, "oldest first")
204 }
205 assert.True(t, list[1].IsGeneral())
206 assert.Equal(t, domain.SideNone, list[1].Side)
207 assert.Equal(t, domain.SideNew, list[0].Side)
208
209 mr, err := mrs.Get(ctx, n, first.Number)
210 require.NoError(t, err)
211 assert.Equal(t, int64(12), mr.CommentSeq)
212
213 c, err := mrs.GetComment(ctx, n, first.Number, 3)
214 require.NoError(t, err)
215 assert.NotZero(t, c.Version)
216 assert.Equal(t, "note", c.Text.String())
217 assert.Equal(t, oid('d'), c.CommitID)
218 for _, id := range []int64{0, 13, -1} {
219 _, err = mrs.GetComment(ctx, n, first.Number, id)
220 assert.ErrorIs(t, err, domain.ErrNotFound)
221 }
222
223 // Another merge request's comments are not listed.
224 second, err := mrs.Create(ctx, newMR(t, n, "t2"))
225 require.NoError(t, err)
226 list, err = mrs.ListComments(ctx, n, second.Number)
227 require.NoError(t, err)
228 assert.Empty(t, list)
229}
230
231func TestSaveCommentPreconditions(t *testing.T) {
232 ctx := context.Background()
233 _, mrs, n := repoWithMRs(t)
234 first, err := mrs.Create(ctx, newMR(t, n, "t"))
235 require.NoError(t, err)
236 mr, err := mrs.Get(ctx, n, first.Number)
237 require.NoError(t, err)
238
239 mr.CommentSeq = 1
240 require.NoError(t, mrs.SaveComment(ctx, mr, newComment(t, 1, "one")))
241
242 // The same id again, from the same stale read: the request moved on.
243 assert.ErrorIs(t, mrs.SaveComment(ctx, mr, newComment(t, 1, "dup")), domain.ErrConflict)
244
245 // A fresh read of the request but a new comment with an existing id.
246 mr, err = mrs.Get(ctx, n, first.Number)
247 require.NoError(t, err)
248 assert.ErrorIs(t, mrs.SaveComment(ctx, mr, newComment(t, 1, "dup")), domain.ErrConflict, "a new comment must not overwrite")
249
250 // Resolving needs the comment version.
251 c, err := mrs.GetComment(ctx, n, first.Number, 1)
252 require.NoError(t, err)
253 stale := c
254 c.Resolved = true
255 require.NoError(t, mrs.SaveComment(ctx, mr, c))
256 mr, err = mrs.Get(ctx, n, first.Number)
257 require.NoError(t, err)
258 stale.Resolved = false
259 assert.ErrorIs(t, mrs.SaveComment(ctx, mr, stale), domain.ErrConflict)
260
261 // An unread request cannot be saved with.
262 assert.ErrorIs(t, mrs.SaveComment(ctx, newMR(t, n, "x"), newComment(t, 2, "n")), domain.ErrConflict)
263}
264
265func TestReadPosition(t *testing.T) {
266 ctx := context.Background()
267 _, mrs, n := repoWithMRs(t)
268 got, err := mrs.GetRead(ctx, n, 1)
269 require.NoError(t, err)
270 assert.True(t, got.IsZero())
271
272 require.NoError(t, mrs.SetRead(ctx, n, 1, oid('a')))
273 require.NoError(t, mrs.SetRead(ctx, n, 1, oid('b')))
274 got, err = mrs.GetRead(ctx, n, 1)
275 require.NoError(t, err)
276 assert.Equal(t, oid('b'), got)
277 got, err = mrs.GetRead(ctx, n, 2)
278 require.NoError(t, err)
279 assert.True(t, got.IsZero())
280 assert.Error(t, mrs.SetRead(ctx, n, 1, domain.ObjectID{}))
281}
282
283// TestDeleteDropsTheMergeRequestsOfTheRepository: deleting a repository drops
284// its merge requests, comments, read positions and number sequence, and only
285// its own; a repository created again under the name starts at !1.
286func TestDeleteDropsTheMergeRequestsOfTheRepository(t *testing.T) {
287 ctx := context.Background()
288 s, mrs, n := repoWithMRs(t)
289 sib := name(t, "proj-2")
290 require.NoError(t, s.Create(ctx, domain.Repo{Name: sib, CreatedAt: created, Visibility: domain.VisibilityPublic}))
291
292 for _, repo := range []domain.RepoName{n, sib} {
293 mr, err := mrs.Create(ctx, newMR(t, repo, "t"))
294 require.NoError(t, err)
295 mr.CommentSeq = 1
296 require.NoError(t, mrs.SaveComment(ctx, mr, newComment(t, 1, "note")))
297 require.NoError(t, mrs.SetRead(ctx, repo, mr.Number, oid('a')))
298 }
299
300 require.NoError(t, s.Delete(ctx, n))
301
302 list, err := mrs.List(ctx, n)
303 require.NoError(t, err)
304 assert.Empty(t, list)
305 comments, err := mrs.ListComments(ctx, n, 1)
306 require.NoError(t, err)
307 assert.Empty(t, comments)
308 read, err := mrs.GetRead(ctx, n, 1)
309 require.NoError(t, err)
310 assert.True(t, read.IsZero())
311
312 list, err = mrs.List(ctx, sib)
313 require.NoError(t, err)
314 require.Len(t, list, 1, "a repository whose name starts like the deleted one keeps its requests")
315 comments, err = mrs.ListComments(ctx, sib, 1)
316 require.NoError(t, err)
317 assert.Len(t, comments, 1)
318
319 require.NoError(t, s.Create(ctx, domain.Repo{Name: n, CreatedAt: created, Visibility: domain.VisibilityPrivate}))
320 again, err := mrs.Create(ctx, newMR(t, n, "fresh"))
321 require.NoError(t, err)
322 assert.Equal(t, int64(1), again.Number)
323}
324
325// TestWritesAfterDeleteLeaveNothing: a comment or a read position written for
326// a repository that was deleted meanwhile is refused, so that a repository
327// created again under the name does not inherit it (GO-002).
328func TestWritesAfterDeleteLeaveNothing(t *testing.T) {
329 ctx := context.Background()
330 s, mrs, n := repoWithMRs(t)
331 mr, err := mrs.Create(ctx, newMR(t, n, "t"))
332 require.NoError(t, err)
333
334 require.NoError(t, s.Delete(ctx, n))
335
336 mr.CommentSeq = 1
337 assert.ErrorIs(t, mrs.SaveComment(ctx, mr, newComment(t, 1, "late")), domain.ErrNotFound)
338 assert.ErrorIs(t, mrs.SetRead(ctx, n, mr.Number, oid('a')), domain.ErrNotFound)
339 _, err = mrs.Create(ctx, newMR(t, n, "late"))
340 assert.ErrorIs(t, err, domain.ErrNotFound)
341
342 require.NoError(t, s.Create(ctx, domain.Repo{Name: n, CreatedAt: created, Visibility: domain.VisibilityPublic}))
343 assertNoMergeRequestKeys(t, mrs, n)
344}
345
346func assertNoMergeRequestKeys(t *testing.T, mrs *kvstore.MergeRequests, n domain.RepoName) {
347 t.Helper()
348 ctx := context.Background()
349 list, err := mrs.List(ctx, n)
350 require.NoError(t, err)
351 assert.Empty(t, list, "merge requests")
352 comments, err := mrs.ListComments(ctx, n, 1)
353 require.NoError(t, err)
354 assert.Empty(t, comments, "comments")
355 read, err := mrs.GetRead(ctx, n, 1)
356 require.NoError(t, err)
357 assert.True(t, read.IsZero(), "read position")
358}
359
360// TestDeleteRacingWithWrites: writes racing with Delete either land before it
361// and are deleted, or are refused (not found after it, or a conflict when the
362// retries run out under this contention); nothing is left for a repository
363// created again under the name. Delete itself may report a conflict under the
364// same contention and is then called again.
365func TestDeleteRacingWithWrites(t *testing.T) {
366 ctx := context.Background()
367 for round := range 20 {
368 s, mrs, n := repoWithMRs(t)
369 mr, err := mrs.Create(ctx, newMR(t, n, "t"))
370 require.NoError(t, err)
371
372 var wg sync.WaitGroup
373 for w := range 4 {
374 wg.Add(1)
375 go func() {
376 defer wg.Done()
377 for i := range 30 {
378 if w%2 == 0 {
379 err := mrs.SetRead(ctx, n, mr.Number, oid('a'))
380 if err != nil {
381 assert.True(t, errors.Is(err, domain.ErrConflict) || errors.Is(err, domain.ErrNotFound), "round %d: %v", round, err)
382 return
383 }
384 continue
385 }
386 cur, err := mrs.Get(ctx, n, mr.Number)
387 if err != nil {
388 return
389 }
390 cur.CommentSeq = int64(w*1000 + i + 1)
391 err = mrs.SaveComment(ctx, cur, newComment(t, cur.CommentSeq, "note"))
392 if err != nil && !assert.True(t, errors.Is(err, domain.ErrConflict) || errors.Is(err, domain.ErrNotFound), "round %d: %v", round, err) {
393 return
394 }
395 }
396 }()
397 }
398 // Delete gives up with a conflict when the writers keep rewriting the
399 // repository key for deleteTries listings in a row; it is documented as
400 // safe to call again, as a caller does.
401 var delErr error
402 for range 20 {
403 if delErr = s.Delete(ctx, n); !errors.Is(delErr, domain.ErrConflict) {
404 break
405 }
406 }
407 wg.Wait()
408 require.NoError(t, delErr)
409
410 require.NoError(t, s.Create(ctx, domain.Repo{Name: n, CreatedAt: created, Visibility: domain.VisibilityPublic}))
411 assertNoMergeRequestKeys(t, mrs, n)
412 again, err := mrs.Create(ctx, newMR(t, n, "fresh"))
413 require.NoError(t, err)
414 assert.Equal(t, int64(1), again.Number)
415 }
416}
417
418func TestDeleteManyMergeRequests(t *testing.T) {
419 ctx := context.Background()
420 s, mrs, n := repoWithMRs(t)
421 for range 1100 {
422 _, err := mrs.Create(ctx, newMR(t, n, "t"))
423 require.NoError(t, err)
424 }
425 require.NoError(t, s.Delete(ctx, n))
426 list, err := mrs.List(ctx, n)
427 require.NoError(t, err)
428 assert.Empty(t, list)
429 _, err = s.Get(ctx, n)
430 assert.ErrorIs(t, err, domain.ErrNotFound)
431}
432
433func TestRepoSettings(t *testing.T) {
434 ctx := context.Background()
435 s := openStore(t, t.TempDir())
436 n := name(t, "proj")
437
438 // Created without settings: they are stored explicitly, with the defaults.
439 require.NoError(t, s.Create(ctx, domain.Repo{Name: n, CreatedAt: created, Visibility: domain.VisibilityPrivate}))
440 got, err := s.Get(ctx, n)
441 require.NoError(t, err)
442 assert.Equal(t, domain.DefaultRepoSettings(), got.Settings)
443
444 got.Settings = domain.RepoSettings{AllowedMergeModes: []domain.MergeMode{domain.MergeModeSquash}, FailingChecksBlock: false}
445 require.NoError(t, s.Update(ctx, got))
446 again, err := s.Get(ctx, n)
447 require.NoError(t, err)
448 assert.Equal(t, got.Settings, again.Settings)
449
450 got.Settings = domain.RepoSettings{AllowedMergeModes: []domain.MergeMode{"rebase"}}
451 assert.ErrorIs(t, s.Update(ctx, got), domain.ErrInvalid)
452}
453
454// TestRepoWithoutStoredSettingsReadsTheDefaults: a repository stored before
455// merge requests existed has no settings field.
456func TestRepoWithoutStoredSettingsReadsTheDefaults(t *testing.T) {
457 ctx := context.Background()
458 dir := t.TempDir()
459 s, err := kvstore.Open(ctx, dir)
460 require.NoError(t, err)
461 require.NoError(t, s.Close(ctx))
462
463 // Write a legacy value straight into the database.
464 db, err := kv.Open(ctx, dir, kv.Options{})
465 require.NoError(t, err)
466 legacy, err := proto.Marshal(&storagepb.Repo{Name: "legacy", Visibility: storagepb.Visibility_VISIBILITY_PUBLIC})
467 require.NoError(t, err)
468 var b kv.Batch
469 b.Put([]byte("repo/legacy"), legacy)
470 _, err = db.Apply(ctx, &b)
471 require.NoError(t, err)
472 require.NoError(t, db.Close(ctx))
473
474 s = openStore(t, dir)
475 got, err := s.Get(ctx, name(t, "legacy"))
476 require.NoError(t, err)
477 assert.Equal(t, domain.DefaultRepoSettings(), got.Settings)
478 list, err := s.List(ctx)
479 require.NoError(t, err)
480 require.Len(t, list, 1)
481 assert.Equal(t, domain.DefaultRepoSettings(), list[0].Settings)
482}
483
484func TestSigningSeed(t *testing.T) {
485 ctx := context.Background()
486 dir := t.TempDir()
487 s := openStore(t, dir)
488 calls := 0
489 gen := func() ([]byte, error) {
490 calls++
491 return []byte(strings.Repeat(string(rune('a'+calls)), kvstore.SigningSeedLen)), nil
492 }
493
494 first, err := s.SigningSeed(ctx, gen, created)
495 require.NoError(t, err)
496 require.Len(t, first, kvstore.SigningSeedLen)
497 second, err := s.SigningSeed(ctx, gen, created)
498 require.NoError(t, err)
499 assert.Equal(t, first, second, "generated once")
500 assert.Equal(t, 1, calls)
501
502 // It survives a restart.
503 require.NoError(t, s.Close(ctx))
504 s2 := openStore(t, dir)
505 third, err := s2.SigningSeed(ctx, gen, created)
506 require.NoError(t, err)
507 assert.Equal(t, first, third)
508 assert.Equal(t, 1, calls)
509}
510
511func TestSigningSeedConcurrentFirstStart(t *testing.T) {
512 ctx := context.Background()
513 s := openStore(t, t.TempDir())
514 var (
515 wg sync.WaitGroup
516 mu sync.Mutex
517 seeds [][]byte
518 n byte
519 )
520 for range 6 {
521 wg.Go(func() {
522 seed, err := s.SigningSeed(ctx, func() ([]byte, error) {
523 mu.Lock()
524 defer mu.Unlock()
525 n++
526 return []byte(strings.Repeat(string(rune('A'+n)), kvstore.SigningSeedLen)), nil
527 }, created)
528 if assert.NoError(t, err) {
529 mu.Lock()
530 seeds = append(seeds, seed)
531 mu.Unlock()
532 }
533 })
534 }
535 wg.Wait()
536 for _, seed := range seeds {
537 assert.Equal(t, seeds[0], seed, "every caller gets the winner's seed")
538 }
539}
540
541func TestSigningSeedRejectsBadGenerators(t *testing.T) {
542 ctx := context.Background()
543 s := openStore(t, t.TempDir())
544 _, err := s.SigningSeed(ctx, func() ([]byte, error) { return []byte("short"), nil }, created)
545 require.Error(t, err)
546 assert.NotContains(t, err.Error(), "short")
547 _, err = s.SigningSeed(ctx, func() ([]byte, error) { return nil, assert.AnError }, created)
548 require.ErrorIs(t, err, assert.AnError)
549}