1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package kvstore_test
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "sync"
10 "testing"
11 "time"
12
13 kv "bnema.dev/ermite-kv"
14 "github.com/stretchr/testify/assert"
15 "github.com/stretchr/testify/require"
16 "google.golang.org/protobuf/proto"
17
18 storagepb "bnema.dev/ermite/gen/ermite/storage"
19 "bnema.dev/ermite/internal/adapters/kvstore"
20 "bnema.dev/ermite/internal/domain"
21)
22
23func openStore(t *testing.T, dir string) *kvstore.Store {
24 t.Helper()
25 ctx := context.Background()
26 s, err := kvstore.Open(ctx, dir)
27 require.NoError(t, err)
28 t.Cleanup(func() { _ = s.Close(ctx) })
29 return s
30}
31
32func name(t *testing.T, s string) domain.RepoName {
33 t.Helper()
34 n, err := domain.ParseRepoName(s)
35 require.NoError(t, err)
36 return n
37}
38
39var created = time.Date(2026, 10, 10, 8, 0, 0, 0, time.UTC)
40
41func TestCreateGetListUpdateDelete(t *testing.T) {
42 ctx := context.Background()
43 s := openStore(t, t.TempDir())
44
45 b := domain.Repo{Name: name(t, "b"), Description: "second", Topics: []string{"cli", "go"}, Visibility: domain.VisibilityPublic, CreatedAt: created, Settings: domain.DefaultRepoSettings()}
46 a := domain.Repo{Name: name(t, "a"), Visibility: domain.VisibilityPrivate, CreatedAt: created, PushedAt: created.Add(time.Hour), ActiveAt: created.Add(2 * time.Hour), Settings: domain.DefaultRepoSettings()}
47 require.NoError(t, s.Create(ctx, b))
48 require.NoError(t, s.Create(ctx, a))
49
50 got, err := s.Get(ctx, b.Name)
51 require.NoError(t, err)
52 assert.Equal(t, b, got)
53 got, err = s.Get(ctx, a.Name)
54 require.NoError(t, err)
55 assert.Equal(t, a, got)
56 assert.True(t, got.PushedAt.Equal(a.PushedAt))
57 assert.True(t, got.ActiveAt.Equal(a.ActiveAt))
58
59 list, err := s.List(ctx)
60 require.NoError(t, err)
61 assert.Equal(t, []domain.Repo{a, b}, list)
62
63 err = s.Create(ctx, b)
64 assert.ErrorIs(t, err, domain.ErrAlreadyExists)
65
66 b.Description = "changed"
67 b.Topics = []string{"forge"}
68 b.Visibility = domain.VisibilityPrivate
69 _, err = s.Mutate(ctx, b.Name, func(r *domain.Repo) (bool, error) { *r = b; return true, nil })
70 require.NoError(t, err)
71 got, err = s.Get(ctx, b.Name)
72 require.NoError(t, err)
73 assert.Equal(t, b, got)
74
75 b.Topics = nil
76 _, err = s.Mutate(ctx, b.Name, func(r *domain.Repo) (bool, error) { *r = b; return true, nil })
77 require.NoError(t, err)
78 got, err = s.Get(ctx, b.Name)
79 require.NoError(t, err)
80 assert.Nil(t, got.Topics, "cleared topics read back as nil")
81
82 require.NoError(t, s.Delete(ctx, b.Name))
83 _, err = s.Get(ctx, b.Name)
84 assert.ErrorIs(t, err, domain.ErrNotFound)
85 require.NoError(t, s.Delete(ctx, b.Name), "deleting a missing repository is not an error")
86 _, err = s.Mutate(ctx, b.Name, func(r *domain.Repo) (bool, error) { return true, nil })
87 assert.ErrorIs(t, err, domain.ErrNotFound)
88
89 _, err = s.Get(ctx, name(t, "never"))
90 assert.ErrorIs(t, err, domain.ErrNotFound)
91}
92
93func TestCreateRejectsInvalidValues(t *testing.T) {
94 ctx := context.Background()
95 s := openStore(t, t.TempDir())
96 n := name(t, "x")
97
98 assert.ErrorIs(t, s.Create(ctx, domain.Repo{Name: n, Visibility: "secret"}), domain.ErrInvalidVisibility)
99 assert.ErrorIs(t, s.Create(ctx, domain.Repo{Visibility: domain.VisibilityPublic}), domain.ErrInvalidName)
100 assert.Error(t, s.Create(ctx, domain.Repo{Name: n, Visibility: domain.VisibilityPublic, Description: "\xff"}))
101 list, err := s.List(ctx)
102 require.NoError(t, err)
103 assert.Empty(t, list)
104}
105
106func TestConcurrentCreateSameName(t *testing.T) {
107 ctx := context.Background()
108 s := openStore(t, t.TempDir())
109 n := name(t, "tools")
110
111 const workers = 16
112 errs := make(chan error, workers)
113 var wg sync.WaitGroup
114 for i := range workers {
115 wg.Add(1)
116 go func() {
117 defer wg.Done()
118 errs <- s.Create(ctx, domain.Repo{Name: n, Description: fmt.Sprint(i), Visibility: domain.VisibilityPrivate, CreatedAt: created})
119 }()
120 }
121 wg.Wait()
122 close(errs)
123 ok, exists := 0, 0
124 for err := range errs {
125 switch {
126 case err == nil:
127 ok++
128 case errors.Is(err, domain.ErrAlreadyExists):
129 exists++
130 default:
131 t.Fatalf("unexpected error: %v", err)
132 }
133 }
134 assert.Equal(t, 1, ok)
135 assert.Equal(t, workers-1, exists)
136}
137
138func TestPersistsAcrossReopen(t *testing.T) {
139 ctx := context.Background()
140 dir := t.TempDir()
141 s, err := kvstore.Open(ctx, dir)
142 require.NoError(t, err)
143 r := domain.Repo{Name: name(t, "keep"), Visibility: domain.VisibilityPublic, CreatedAt: created, Settings: domain.DefaultRepoSettings()}
144 require.NoError(t, s.Create(ctx, r))
145 require.NoError(t, s.Close(ctx))
146
147 s2 := openStore(t, dir)
148 got, err := s2.Get(ctx, r.Name)
149 require.NoError(t, err)
150 assert.Equal(t, r, got)
151}
152
153func TestCheck(t *testing.T) {
154 ctx := context.Background()
155 s, err := kvstore.Open(ctx, t.TempDir())
156 require.NoError(t, err)
157 require.NoError(t, s.Check(ctx))
158
159 // The caller's own cancellation is reported, not mistaken for health.
160 cancelled, cancel := context.WithCancel(ctx)
161 cancel()
162 assert.ErrorIs(t, s.Check(cancelled), context.Canceled)
163
164 require.NoError(t, s.Close(ctx))
165 assert.ErrorIs(t, s.Check(ctx), kv.ErrClosed)
166}
167
168func TestOpenLockedDirectory(t *testing.T) {
169 ctx := context.Background()
170 dir := t.TempDir()
171 _ = openStore(t, dir)
172 _, err := kvstore.Open(ctx, dir)
173 assert.ErrorIs(t, err, kv.ErrLocked)
174}
175
176// putRaw writes a value with ermite-kv directly, bypassing the store.
177func putRaw(t *testing.T, dir, key string, val []byte) {
178 t.Helper()
179 ctx := context.Background()
180 db, err := kv.Open(ctx, dir, kv.Options{})
181 require.NoError(t, err)
182 var b kv.Batch
183 b.Put([]byte(key), val)
184 _, err = db.Apply(ctx, &b)
185 require.NoError(t, err)
186 require.NoError(t, db.Close(ctx))
187}
188
189func readSchema(t *testing.T, dir string) uint32 {
190 t.Helper()
191 ctx := context.Background()
192 db, err := kv.Open(ctx, dir, kv.Options{})
193 require.NoError(t, err)
194 defer func() { _ = db.Close(ctx) }()
195 item, ok := db.Get(ctx, []byte("meta/schema"))
196 require.True(t, ok, "meta/schema missing")
197 var sv storagepb.SchemaVersion
198 require.NoError(t, proto.Unmarshal(item.Value, &sv))
199 return sv.GetVersion()
200}
201
202func TestSchema(t *testing.T) {
203 ctx := context.Background()
204
205 t.Run("missing is written as the current version", func(t *testing.T) {
206 dir := t.TempDir()
207 s, err := kvstore.Open(ctx, dir)
208 require.NoError(t, err)
209 require.NoError(t, s.Close(ctx))
210 assert.Equal(t, uint32(kvstore.SchemaVersion), readSchema(t, dir))
211 })
212
213 t.Run("current is accepted and kept", func(t *testing.T) {
214 dir := t.TempDir()
215 s, err := kvstore.Open(ctx, dir)
216 require.NoError(t, err)
217 require.NoError(t, s.Close(ctx))
218 s, err = kvstore.Open(ctx, dir)
219 require.NoError(t, err)
220 require.NoError(t, s.Close(ctx))
221 assert.Equal(t, uint32(kvstore.SchemaVersion), readSchema(t, dir))
222 })
223
224 t.Run("version 1 is upgraded to the current one, its data kept", func(t *testing.T) {
225 dir := t.TempDir()
226 putRaw(t, dir, "meta/schema", mustProto(t, &storagepb.SchemaVersion{Version: 1}))
227 putRaw(t, dir, "repo/old", mustProto(t, &storagepb.Repo{Name: "old", Visibility: storagepb.Visibility_VISIBILITY_PUBLIC}))
228 s, err := kvstore.Open(ctx, dir)
229 require.NoError(t, err)
230 got, err := s.Get(ctx, name(t, "old"))
231 require.NoError(t, err)
232 assert.Equal(t, "old", got.Name.String())
233 require.NoError(t, s.Close(ctx))
234 assert.Equal(t, uint32(kvstore.SchemaVersion), readSchema(t, dir))
235 })
236
237 t.Run("version 2 is upgraded to the current one, its runs kept", func(t *testing.T) {
238 dir := t.TempDir()
239 putRaw(t, dir, "meta/schema", mustProto(t, &storagepb.SchemaVersion{Version: 2}))
240 putRaw(t, dir, "repo/old", mustProto(t, &storagepb.Repo{Name: "old", Visibility: storagepb.Visibility_VISIBILITY_PUBLIC}))
241 // A run written by a schema 2 build (no head sha), and a retained pull_request
242 // run that has one: both must read back after the upgrade.
243 putRaw(t, dir, "run/old/00000000000000000001", mustProto(t, &storagepb.CIRun{
244 Repo: "old", Id: 1, Number: 1, Attempt: 1, Event: "push", Ref: "refs/heads/main", Sha: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
245 Status: storagepb.CIStatus_CI_STATUS_COMPLETED,
246 }))
247 putRaw(t, dir, "run/old/00000000000000000002", mustProto(t, &storagepb.CIRun{
248 Repo: "old", Id: 2, Number: 2, Attempt: 1, Event: "pull_request", Ref: "refs/pull/1/merge",
249 Sha: "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", HeadSha: "cccccccccccccccccccccccccccccccccccccccc",
250 Status: storagepb.CIStatus_CI_STATUS_COMPLETED,
251 }))
252 s, err := kvstore.Open(ctx, dir)
253 require.NoError(t, err)
254 got, err := s.Get(ctx, name(t, "old"))
255 require.NoError(t, err)
256 assert.Equal(t, "old", got.Name.String())
257 old, err := s.CI().GetRun(ctx, name(t, "old"), 1)
258 require.NoError(t, err)
259 assert.Equal(t, "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", old.Sha)
260 assert.Empty(t, old.HeadSha)
261 assert.Equal(t, old.Sha, old.CheckSha(), "a run without a head sha keeps its checks on its commit")
262 pr, err := s.CI().GetRun(ctx, name(t, "old"), 2)
263 require.NoError(t, err)
264 assert.Equal(t, "pull_request", pr.Event)
265 assert.Equal(t, "cccccccccccccccccccccccccccccccccccccccc", pr.HeadSha)
266 assert.Equal(t, pr.HeadSha, pr.CheckSha(), "the checks of a pull_request run are on the head")
267 require.NoError(t, s.Close(ctx))
268 assert.Equal(t, uint32(kvstore.SchemaVersion), readSchema(t, dir))
269 })
270
271 t.Run("version 2 is upgraded to the current one, its data kept", func(t *testing.T) {
272 dir := t.TempDir()
273 putRaw(t, dir, "meta/schema", mustProto(t, &storagepb.SchemaVersion{Version: 2}))
274 putRaw(t, dir, "repo/old", mustProto(t, &storagepb.Repo{Name: "old", Visibility: storagepb.Visibility_VISIBILITY_PUBLIC}))
275 s, err := kvstore.Open(ctx, dir)
276 require.NoError(t, err)
277 _, err = s.Get(ctx, name(t, "old"))
278 require.NoError(t, err)
279 require.NoError(t, s.Close(ctx))
280 assert.Equal(t, uint32(kvstore.SchemaVersion), readSchema(t, dir))
281 })
282
283 // Downgrade guard: once a build with reusable workflows has opened the data
284 // (and may have stored calls), a build that knows schema 2 only must refuse
285 // it; it does so for any version above its own (ErrSchemaTooNew), which is
286 // the version 3 this build writes.
287 t.Run("data written by this build is above schema 2", func(t *testing.T) {
288 dir := t.TempDir()
289 s, err := kvstore.Open(ctx, dir)
290 require.NoError(t, err)
291 require.NoError(t, s.Close(ctx))
292 assert.Greater(t, readSchema(t, dir), uint32(2), "a build that supports 2 refuses it")
293 })
294
295 t.Run("newer stops startup and releases the lock", func(t *testing.T) {
296 dir := t.TempDir()
297 putRaw(t, dir, "meta/schema", mustProto(t, &storagepb.SchemaVersion{Version: kvstore.SchemaVersion + 1}))
298 _, err := kvstore.Open(ctx, dir)
299 require.ErrorIs(t, err, kvstore.ErrSchemaTooNew)
300 assert.Equal(t, uint32(kvstore.SchemaVersion+1), readSchema(t, dir), "the lock was released and the value kept")
301 })
302
303 t.Run("garbage and zero are errors", func(t *testing.T) {
304 dir := t.TempDir()
305 putRaw(t, dir, "meta/schema", []byte{0xff, 0xff, 0xff})
306 _, err := kvstore.Open(ctx, dir)
307 assert.Error(t, err)
308
309 dir = t.TempDir()
310 putRaw(t, dir, "meta/schema", mustProto(t, &storagepb.SchemaVersion{Version: 0}))
311 _, err = kvstore.Open(ctx, dir)
312 assert.Error(t, err)
313 })
314}
315
316func mustProto(t *testing.T, m proto.Message) []byte {
317 t.Helper()
318 b, err := proto.Marshal(m)
319 require.NoError(t, err)
320 return b
321}
322
323func TestUnspecifiedVisibilityReadsPrivate(t *testing.T) {
324 ctx := context.Background()
325 dir := t.TempDir()
326 putRaw(t, dir, "repo/odd", mustProto(t, &storagepb.Repo{Name: "odd"})) // visibility unset
327 putRaw(t, dir, "repo/future", mustProto(t, &storagepb.Repo{Name: "future", Visibility: storagepb.Visibility(42)}))
328 s := openStore(t, dir)
329
330 for _, n := range []string{"odd", "future"} {
331 got, err := s.Get(ctx, name(t, n))
332 require.NoError(t, err)
333 assert.Equal(t, domain.VisibilityPrivate, got.Visibility, n)
334 }
335}
336
337func TestCorruptRepoValue(t *testing.T) {
338 ctx := context.Background()
339 dir := t.TempDir()
340 putRaw(t, dir, "repo/bad", []byte{0xff, 0xff})
341 putRaw(t, dir, "repo/mismatch", mustProto(t, &storagepb.Repo{Name: "Not Valid"}))
342 s := openStore(t, dir)
343
344 _, err := s.Get(ctx, name(t, "bad"))
345 assert.Error(t, err)
346 _, err = s.Get(ctx, name(t, "mismatch"))
347 assert.ErrorIs(t, err, domain.ErrInvalidName)
348 _, err = s.List(ctx)
349 assert.Error(t, err)
350}
351
352func TestRepoWithoutTopicsReadsNil(t *testing.T) {
353 ctx := context.Background()
354 dir := t.TempDir()
355 putRaw(t, dir, "repo/old", mustProto(t, &storagepb.Repo{Name: "old", Visibility: storagepb.Visibility_VISIBILITY_PUBLIC}))
356 putRaw(t, dir, "repo/new", mustProto(t, &storagepb.Repo{Name: "new", Visibility: storagepb.Visibility_VISIBILITY_PUBLIC, Topics: []string{"a", "b"}}))
357 s := openStore(t, dir)
358
359 old, err := s.Get(ctx, name(t, "old"))
360 require.NoError(t, err)
361 assert.Nil(t, old.Topics)
362 got, err := s.Get(ctx, name(t, "new"))
363 require.NoError(t, err)
364 assert.Equal(t, []string{"a", "b"}, got.Topics)
365}
366
367func TestSchema6TokenUpgrade(t *testing.T) {
368 ctx := context.Background()
369 for _, from := range []uint32{3, 4, 5} {
370 t.Run(fmt.Sprint(from), func(t *testing.T) {
371 dir := t.TempDir()
372 putRaw(t, dir, "meta/schema", mustProto(t, &storagepb.SchemaVersion{Version: from}))
373 all := domain.AllScopes()
374 scopes := make([]string, 0, len(all)-1)
375 for _, sc := range all {
376 if sc != domain.ScopeReleaseWrite {
377 scopes = append(scopes, sc.String())
378 }
379 }
380 for _, tc := range []struct {
381 id string
382 scopes []string
383 repos, prefixes []string
384 upgraded bool
385 }{
386 {id: "full", scopes: scopes, upgraded: true},
387 {id: "partial", scopes: []string{"repo:read"}},
388 {id: "missing-review", scopes: func() []string {
389 var out []string
390 for _, s := range scopes {
391 if s != "mr:review" {
392 out = append(out, s)
393 }
394 }
395 return out
396 }(), upgraded: from < 4},
397 {id: "repo", scopes: scopes, repos: []string{"a"}},
398 {id: "branch", scopes: scopes, prefixes: []string{"bot/"}},
399 } {
400 putRaw(t, dir, "token/"+tc.id, mustProto(t, &storagepb.Token{Id: tc.id, Scopes: tc.scopes, Repos: tc.repos, BranchPrefixes: tc.prefixes}))
401 }
402 s, err := kvstore.Open(ctx, dir)
403 require.NoError(t, err)
404 require.NoError(t, s.Close(ctx))
405 assert.Equal(t, uint32(kvstore.SchemaVersion), readSchema(t, dir))
406 db, err := kv.Open(ctx, dir, kv.Options{})
407 require.NoError(t, err)
408 defer db.Close(ctx)
409 for _, id := range []string{"full", "partial", "missing-review", "repo", "branch"} {
410 item, ok := db.Get(ctx, []byte("token/"+id))
411 require.True(t, ok)
412 var pb storagepb.Token
413 require.NoError(t, proto.Unmarshal(item.Value, &pb))
414 if id == "full" || (id == "missing-review" && from < 4) {
415 assert.Contains(t, pb.Scopes, "release:write")
416 } else {
417 assert.NotContains(t, pb.Scopes, "release:write")
418 }
419 }
420 })
421 }
422}