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), 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
58 list, err := s.List(ctx)
59 require.NoError(t, err)
60 assert.Equal(t, []domain.Repo{a, b}, list)
61
62 err = s.Create(ctx, b)
63 assert.ErrorIs(t, err, domain.ErrAlreadyExists)
64
65 b.Description = "changed"
66 b.Topics = []string{"forge"}
67 b.Visibility = domain.VisibilityPrivate
68 require.NoError(t, s.Update(ctx, b))
69 got, err = s.Get(ctx, b.Name)
70 require.NoError(t, err)
71 assert.Equal(t, b, got)
72
73 b.Topics = nil
74 require.NoError(t, s.Update(ctx, b))
75 got, err = s.Get(ctx, b.Name)
76 require.NoError(t, err)
77 assert.Nil(t, got.Topics, "cleared topics read back as nil")
78
79 require.NoError(t, s.Delete(ctx, b.Name))
80 _, err = s.Get(ctx, b.Name)
81 assert.ErrorIs(t, err, domain.ErrNotFound)
82 require.NoError(t, s.Delete(ctx, b.Name), "deleting a missing repository is not an error")
83 assert.ErrorIs(t, s.Update(ctx, b), domain.ErrNotFound)
84
85 _, err = s.Get(ctx, name(t, "never"))
86 assert.ErrorIs(t, err, domain.ErrNotFound)
87}
88
89func TestCreateRejectsInvalidValues(t *testing.T) {
90 ctx := context.Background()
91 s := openStore(t, t.TempDir())
92 n := name(t, "x")
93
94 assert.ErrorIs(t, s.Create(ctx, domain.Repo{Name: n, Visibility: "secret"}), domain.ErrInvalidVisibility)
95 assert.ErrorIs(t, s.Create(ctx, domain.Repo{Visibility: domain.VisibilityPublic}), domain.ErrInvalidName)
96 assert.Error(t, s.Create(ctx, domain.Repo{Name: n, Visibility: domain.VisibilityPublic, Description: "\xff"}))
97 list, err := s.List(ctx)
98 require.NoError(t, err)
99 assert.Empty(t, list)
100}
101
102func TestConcurrentCreateSameName(t *testing.T) {
103 ctx := context.Background()
104 s := openStore(t, t.TempDir())
105 n := name(t, "tools")
106
107 const workers = 16
108 errs := make(chan error, workers)
109 var wg sync.WaitGroup
110 for i := range workers {
111 wg.Add(1)
112 go func() {
113 defer wg.Done()
114 errs <- s.Create(ctx, domain.Repo{Name: n, Description: fmt.Sprint(i), Visibility: domain.VisibilityPrivate, CreatedAt: created})
115 }()
116 }
117 wg.Wait()
118 close(errs)
119 ok, exists := 0, 0
120 for err := range errs {
121 switch {
122 case err == nil:
123 ok++
124 case errors.Is(err, domain.ErrAlreadyExists):
125 exists++
126 default:
127 t.Fatalf("unexpected error: %v", err)
128 }
129 }
130 assert.Equal(t, 1, ok)
131 assert.Equal(t, workers-1, exists)
132}
133
134func TestPersistsAcrossReopen(t *testing.T) {
135 ctx := context.Background()
136 dir := t.TempDir()
137 s, err := kvstore.Open(ctx, dir)
138 require.NoError(t, err)
139 r := domain.Repo{Name: name(t, "keep"), Visibility: domain.VisibilityPublic, CreatedAt: created, Settings: domain.DefaultRepoSettings()}
140 require.NoError(t, s.Create(ctx, r))
141 require.NoError(t, s.Close(ctx))
142
143 s2 := openStore(t, dir)
144 got, err := s2.Get(ctx, r.Name)
145 require.NoError(t, err)
146 assert.Equal(t, r, got)
147}
148
149func TestCheck(t *testing.T) {
150 ctx := context.Background()
151 s, err := kvstore.Open(ctx, t.TempDir())
152 require.NoError(t, err)
153 require.NoError(t, s.Check(ctx))
154
155 // The caller's own cancellation is reported, not mistaken for health.
156 cancelled, cancel := context.WithCancel(ctx)
157 cancel()
158 assert.ErrorIs(t, s.Check(cancelled), context.Canceled)
159
160 require.NoError(t, s.Close(ctx))
161 assert.ErrorIs(t, s.Check(ctx), kv.ErrClosed)
162}
163
164func TestOpenLockedDirectory(t *testing.T) {
165 ctx := context.Background()
166 dir := t.TempDir()
167 _ = openStore(t, dir)
168 _, err := kvstore.Open(ctx, dir)
169 assert.ErrorIs(t, err, kv.ErrLocked)
170}
171
172// putRaw writes a value with ermite-kv directly, bypassing the store.
173func putRaw(t *testing.T, dir, key string, val []byte) {
174 t.Helper()
175 ctx := context.Background()
176 db, err := kv.Open(ctx, dir, kv.Options{})
177 require.NoError(t, err)
178 var b kv.Batch
179 b.Put([]byte(key), val)
180 _, err = db.Apply(ctx, &b)
181 require.NoError(t, err)
182 require.NoError(t, db.Close(ctx))
183}
184
185func readSchema(t *testing.T, dir string) uint32 {
186 t.Helper()
187 ctx := context.Background()
188 db, err := kv.Open(ctx, dir, kv.Options{})
189 require.NoError(t, err)
190 defer func() { _ = db.Close(ctx) }()
191 item, ok := db.Get(ctx, []byte("meta/schema"))
192 require.True(t, ok, "meta/schema missing")
193 var sv storagepb.SchemaVersion
194 require.NoError(t, proto.Unmarshal(item.Value, &sv))
195 return sv.GetVersion()
196}
197
198func TestSchema(t *testing.T) {
199 ctx := context.Background()
200
201 t.Run("missing is written as the current version", func(t *testing.T) {
202 dir := t.TempDir()
203 s, err := kvstore.Open(ctx, dir)
204 require.NoError(t, err)
205 require.NoError(t, s.Close(ctx))
206 assert.Equal(t, uint32(kvstore.SchemaVersion), readSchema(t, dir))
207 })
208
209 t.Run("current is accepted and kept", func(t *testing.T) {
210 dir := t.TempDir()
211 s, err := kvstore.Open(ctx, dir)
212 require.NoError(t, err)
213 require.NoError(t, s.Close(ctx))
214 s, err = kvstore.Open(ctx, dir)
215 require.NoError(t, err)
216 require.NoError(t, s.Close(ctx))
217 assert.Equal(t, uint32(kvstore.SchemaVersion), readSchema(t, dir))
218 })
219
220 t.Run("version 1 is upgraded to the current one, its data kept", func(t *testing.T) {
221 dir := t.TempDir()
222 putRaw(t, dir, "meta/schema", mustProto(t, &storagepb.SchemaVersion{Version: 1}))
223 putRaw(t, dir, "repo/old", mustProto(t, &storagepb.Repo{Name: "old", Visibility: storagepb.Visibility_VISIBILITY_PUBLIC}))
224 s, err := kvstore.Open(ctx, dir)
225 require.NoError(t, err)
226 got, err := s.Get(ctx, name(t, "old"))
227 require.NoError(t, err)
228 assert.Equal(t, "old", got.Name.String())
229 require.NoError(t, s.Close(ctx))
230 assert.Equal(t, uint32(kvstore.SchemaVersion), readSchema(t, dir))
231 })
232
233 t.Run("version 2 is upgraded to 3, its runs kept", func(t *testing.T) {
234 dir := t.TempDir()
235 putRaw(t, dir, "meta/schema", mustProto(t, &storagepb.SchemaVersion{Version: 2}))
236 putRaw(t, dir, "repo/old", mustProto(t, &storagepb.Repo{Name: "old", Visibility: storagepb.Visibility_VISIBILITY_PUBLIC}))
237 // A run written by a schema 2 build (no head sha), and a retained pull_request
238 // run that has one: both must read back after the upgrade.
239 putRaw(t, dir, "run/old/00000000000000000001", mustProto(t, &storagepb.CIRun{
240 Repo: "old", Id: 1, Number: 1, Attempt: 1, Event: "push", Ref: "refs/heads/main", Sha: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
241 Status: storagepb.CIStatus_CI_STATUS_COMPLETED,
242 }))
243 putRaw(t, dir, "run/old/00000000000000000002", mustProto(t, &storagepb.CIRun{
244 Repo: "old", Id: 2, Number: 2, Attempt: 1, Event: "pull_request", Ref: "refs/pull/1/merge",
245 Sha: "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", HeadSha: "cccccccccccccccccccccccccccccccccccccccc",
246 Status: storagepb.CIStatus_CI_STATUS_COMPLETED,
247 }))
248 s, err := kvstore.Open(ctx, dir)
249 require.NoError(t, err)
250 got, err := s.Get(ctx, name(t, "old"))
251 require.NoError(t, err)
252 assert.Equal(t, "old", got.Name.String())
253 old, err := s.CI().GetRun(ctx, name(t, "old"), 1)
254 require.NoError(t, err)
255 assert.Equal(t, "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", old.Sha)
256 assert.Empty(t, old.HeadSha)
257 assert.Equal(t, old.Sha, old.CheckSha(), "a run without a head sha keeps its checks on its commit")
258 pr, err := s.CI().GetRun(ctx, name(t, "old"), 2)
259 require.NoError(t, err)
260 assert.Equal(t, "pull_request", pr.Event)
261 assert.Equal(t, "cccccccccccccccccccccccccccccccccccccccc", pr.HeadSha)
262 assert.Equal(t, pr.HeadSha, pr.CheckSha(), "the checks of a pull_request run are on the head")
263 require.NoError(t, s.Close(ctx))
264 assert.Equal(t, uint32(3), readSchema(t, dir))
265 })
266
267 t.Run("version 2 is upgraded to 3, its data kept", func(t *testing.T) {
268 dir := t.TempDir()
269 putRaw(t, dir, "meta/schema", mustProto(t, &storagepb.SchemaVersion{Version: 2}))
270 putRaw(t, dir, "repo/old", mustProto(t, &storagepb.Repo{Name: "old", Visibility: storagepb.Visibility_VISIBILITY_PUBLIC}))
271 s, err := kvstore.Open(ctx, dir)
272 require.NoError(t, err)
273 _, err = s.Get(ctx, name(t, "old"))
274 require.NoError(t, err)
275 require.NoError(t, s.Close(ctx))
276 assert.Equal(t, uint32(3), readSchema(t, dir))
277 })
278
279 // Downgrade guard: once a build with reusable workflows has opened the data
280 // (and may have stored calls), a build that knows schema 2 only must refuse
281 // it; it does so for any version above its own (ErrSchemaTooNew), which is
282 // the version 3 this build writes.
283 t.Run("data written by this build is above schema 2", func(t *testing.T) {
284 dir := t.TempDir()
285 s, err := kvstore.Open(ctx, dir)
286 require.NoError(t, err)
287 require.NoError(t, s.Close(ctx))
288 assert.Greater(t, readSchema(t, dir), uint32(2), "a build that supports 2 refuses it")
289 })
290
291 t.Run("newer stops startup and releases the lock", func(t *testing.T) {
292 dir := t.TempDir()
293 putRaw(t, dir, "meta/schema", mustProto(t, &storagepb.SchemaVersion{Version: kvstore.SchemaVersion + 1}))
294 _, err := kvstore.Open(ctx, dir)
295 require.ErrorIs(t, err, kvstore.ErrSchemaTooNew)
296 assert.Equal(t, uint32(kvstore.SchemaVersion+1), readSchema(t, dir), "the lock was released and the value kept")
297 })
298
299 t.Run("garbage and zero are errors", func(t *testing.T) {
300 dir := t.TempDir()
301 putRaw(t, dir, "meta/schema", []byte{0xff, 0xff, 0xff})
302 _, err := kvstore.Open(ctx, dir)
303 assert.Error(t, err)
304
305 dir = t.TempDir()
306 putRaw(t, dir, "meta/schema", mustProto(t, &storagepb.SchemaVersion{Version: 0}))
307 _, err = kvstore.Open(ctx, dir)
308 assert.Error(t, err)
309 })
310}
311
312func mustProto(t *testing.T, m proto.Message) []byte {
313 t.Helper()
314 b, err := proto.Marshal(m)
315 require.NoError(t, err)
316 return b
317}
318
319func TestUnspecifiedVisibilityReadsPrivate(t *testing.T) {
320 ctx := context.Background()
321 dir := t.TempDir()
322 putRaw(t, dir, "repo/odd", mustProto(t, &storagepb.Repo{Name: "odd"})) // visibility unset
323 putRaw(t, dir, "repo/future", mustProto(t, &storagepb.Repo{Name: "future", Visibility: storagepb.Visibility(42)}))
324 s := openStore(t, dir)
325
326 for _, n := range []string{"odd", "future"} {
327 got, err := s.Get(ctx, name(t, n))
328 require.NoError(t, err)
329 assert.Equal(t, domain.VisibilityPrivate, got.Visibility, n)
330 }
331}
332
333func TestCorruptRepoValue(t *testing.T) {
334 ctx := context.Background()
335 dir := t.TempDir()
336 putRaw(t, dir, "repo/bad", []byte{0xff, 0xff})
337 putRaw(t, dir, "repo/mismatch", mustProto(t, &storagepb.Repo{Name: "Not Valid"}))
338 s := openStore(t, dir)
339
340 _, err := s.Get(ctx, name(t, "bad"))
341 assert.Error(t, err)
342 _, err = s.Get(ctx, name(t, "mismatch"))
343 assert.ErrorIs(t, err, domain.ErrInvalidName)
344 _, err = s.List(ctx)
345 assert.Error(t, err)
346}
347
348func TestRepoWithoutTopicsReadsNil(t *testing.T) {
349 ctx := context.Background()
350 dir := t.TempDir()
351 putRaw(t, dir, "repo/old", mustProto(t, &storagepb.Repo{Name: "old", Visibility: storagepb.Visibility_VISIBILITY_PUBLIC}))
352 putRaw(t, dir, "repo/new", mustProto(t, &storagepb.Repo{Name: "new", Visibility: storagepb.Visibility_VISIBILITY_PUBLIC, Topics: []string{"a", "b"}}))
353 s := openStore(t, dir)
354
355 old, err := s.Get(ctx, name(t, "old"))
356 require.NoError(t, err)
357 assert.Nil(t, old.Topics)
358 got, err := s.Get(ctx, name(t, "new"))
359 require.NoError(t, err)
360 assert.Equal(t, []string{"a", "b"}, got.Topics)
361}