1package cache
2
3import (
4 "bytes"
5 "encoding/gob"
6 "fmt"
7 "sync"
8
9 "github.com/pkg/errors"
10
11 "github.com/MichaelMure/git-bug/entities/identity"
12 "github.com/MichaelMure/git-bug/entity"
13 "github.com/MichaelMure/git-bug/repository"
14)
15
16type Excerpt interface {
17 Id() entity.Id
18 setId(id entity.Id)
19}
20
21type CacheEntity interface {
22 Id() entity.Id
23 NeedCommit() bool
24}
25
26type getUserIdentityFunc func() (*IdentityCache, error)
27
28// Actions expose a number of action functions on Entities, to give upper layers (cache) a way to normalize interactions.
29// Note: ideally this wouldn't exist, the cache layer would assume that everything is an entity/dag, and directly use the
30// functions from this package, but right now identities are not using that framework.
31type Actions[EntityT entity.Interface] struct {
32 ReadWithResolver func(repo repository.ClockedRepo, resolvers entity.Resolvers, id entity.Id) (EntityT, error)
33 ReadAllWithResolver func(repo repository.ClockedRepo, resolvers entity.Resolvers) <-chan entity.StreamedEntity[EntityT]
34 Remove func(repo repository.ClockedRepo, id entity.Id) error
35 MergeAll func(repo repository.ClockedRepo, resolvers entity.Resolvers, remote string, mergeAuthor identity.Interface) <-chan entity.MergeResult
36}
37
38var _ cacheMgmt = &SubCache[entity.Interface, Excerpt, CacheEntity]{}
39
40type SubCache[EntityT entity.Interface, ExcerptT Excerpt, CacheT CacheEntity] struct {
41 repo repository.ClockedRepo
42 resolvers func() entity.Resolvers
43
44 getUserIdentity getUserIdentityFunc
45 makeCached func(entity EntityT, entityUpdated func(id entity.Id) error) CacheT
46 makeExcerpt func(CacheT) ExcerptT
47 makeIndexData func(CacheT) []string
48 actions Actions[EntityT]
49
50 typename string
51 namespace string
52 version uint
53 maxLoaded int
54
55 mu sync.RWMutex
56 excerpts map[entity.Id]ExcerptT
57 cached map[entity.Id]CacheT
58 lru *lruIdCache
59}
60
61func NewSubCache[EntityT entity.Interface, ExcerptT Excerpt, CacheT CacheEntity](
62 repo repository.ClockedRepo,
63 resolvers func() entity.Resolvers, getUserIdentity getUserIdentityFunc,
64 makeCached func(entity EntityT, entityUpdated func(id entity.Id) error) CacheT,
65 makeExcerpt func(CacheT) ExcerptT,
66 makeIndexData func(CacheT) []string,
67 actions Actions[EntityT],
68 typename, namespace string,
69 version uint, maxLoaded int) *SubCache[EntityT, ExcerptT, CacheT] {
70 return &SubCache[EntityT, ExcerptT, CacheT]{
71 repo: repo,
72 resolvers: resolvers,
73 getUserIdentity: getUserIdentity,
74 makeCached: makeCached,
75 makeExcerpt: makeExcerpt,
76 makeIndexData: makeIndexData,
77 actions: actions,
78 typename: typename,
79 namespace: namespace,
80 version: version,
81 maxLoaded: maxLoaded,
82 excerpts: make(map[entity.Id]ExcerptT),
83 cached: make(map[entity.Id]CacheT),
84 lru: newLRUIdCache(),
85 }
86}
87
88func (sc *SubCache[EntityT, ExcerptT, CacheT]) Typename() string {
89 return sc.typename
90}
91
92// Load will try to read from the disk the entity cache file
93func (sc *SubCache[EntityT, ExcerptT, CacheT]) Load() error {
94 sc.mu.Lock()
95 defer sc.mu.Unlock()
96
97 f, err := sc.repo.LocalStorage().Open(sc.namespace + "-file")
98 if err != nil {
99 return err
100 }
101
102 decoder := gob.NewDecoder(f)
103
104 aux := struct {
105 Version uint
106 Excerpts map[entity.Id]ExcerptT
107 }{}
108
109 err = decoder.Decode(&aux)
110 if err != nil {
111 return err
112 }
113
114 if aux.Version != sc.version {
115 return fmt.Errorf("unknown %s cache format version %v", sc.namespace, aux.Version)
116 }
117
118 // the id is not serialized in the excerpt itself (non-exported field in go, long story ...),
119 // so we fix it here, which doubles as enforcing coherency.
120 for id, excerpt := range aux.Excerpts {
121 excerpt.setId(id)
122 }
123
124 sc.excerpts = aux.Excerpts
125
126 index, err := sc.repo.GetIndex(sc.typename)
127 if err != nil {
128 return err
129 }
130
131 // simple heuristic to detect a mismatch between the index and the entities
132 count, err := index.DocCount()
133 if err != nil {
134 return err
135 }
136 if count != uint64(len(sc.excerpts)) {
137 return fmt.Errorf("count mismatch between bleve and %s excerpts", sc.namespace)
138 }
139
140 return nil
141}
142
143// Write will serialize on disk the entity cache file
144func (sc *SubCache[EntityT, ExcerptT, CacheT]) write() error {
145 sc.mu.RLock()
146 defer sc.mu.RUnlock()
147
148 var data bytes.Buffer
149
150 aux := struct {
151 Version uint
152 Excerpts map[entity.Id]ExcerptT
153 }{
154 Version: sc.version,
155 Excerpts: sc.excerpts,
156 }
157
158 encoder := gob.NewEncoder(&data)
159
160 err := encoder.Encode(aux)
161 if err != nil {
162 return err
163 }
164
165 f, err := sc.repo.LocalStorage().Create(sc.namespace + "-file")
166 if err != nil {
167 return err
168 }
169
170 _, err = f.Write(data.Bytes())
171 if err != nil {
172 return err
173 }
174
175 return f.Close()
176}
177
178func (sc *SubCache[EntityT, ExcerptT, CacheT]) Build() error {
179 sc.excerpts = make(map[entity.Id]ExcerptT)
180
181 allEntities := sc.actions.ReadAllWithResolver(sc.repo, sc.resolvers())
182
183 index, err := sc.repo.GetIndex(sc.typename)
184 if err != nil {
185 return err
186 }
187
188 // wipe the index just to be sure
189 err = index.Clear()
190 if err != nil {
191 return err
192 }
193
194 indexer, indexEnd := index.IndexBatch()
195
196 for e := range allEntities {
197 if e.Err != nil {
198 return e.Err
199 }
200
201 // TODO: doesn't actually record in cache, should we?
202 cached := sc.makeCached(e.Entity, sc.entityUpdated)
203 sc.excerpts[e.Entity.Id()] = sc.makeExcerpt(cached)
204
205 indexData := sc.makeIndexData(cached)
206 if err := indexer(e.Entity.Id().String(), indexData); err != nil {
207 return err
208 }
209 }
210
211 err = indexEnd()
212 if err != nil {
213 return err
214 }
215
216 err = sc.write()
217 if err != nil {
218 return err
219 }
220
221 return nil
222}
223
224func (sc *SubCache[EntityT, ExcerptT, CacheT]) SetCacheSize(size int) {
225 sc.maxLoaded = size
226 sc.evictIfNeeded()
227}
228
229func (sc *SubCache[EntityT, ExcerptT, CacheT]) Close() error {
230 sc.mu.Lock()
231 defer sc.mu.Unlock()
232 sc.excerpts = nil
233 sc.cached = make(map[entity.Id]CacheT)
234 return nil
235}
236
237// AllIds return all known bug ids
238func (sc *SubCache[EntityT, ExcerptT, CacheT]) AllIds() []entity.Id {
239 sc.mu.RLock()
240 defer sc.mu.RUnlock()
241
242 result := make([]entity.Id, len(sc.excerpts))
243
244 i := 0
245 for _, excerpt := range sc.excerpts {
246 result[i] = excerpt.Id()
247 i++
248 }
249
250 return result
251}
252
253// Resolve retrieve an entity matching the exact given id
254func (sc *SubCache[EntityT, ExcerptT, CacheT]) Resolve(id entity.Id) (CacheT, error) {
255 sc.mu.RLock()
256 cached, ok := sc.cached[id]
257 if ok {
258 sc.lru.Get(id)
259 sc.mu.RUnlock()
260 return cached, nil
261 }
262 sc.mu.RUnlock()
263
264 e, err := sc.actions.ReadWithResolver(sc.repo, sc.resolvers(), id)
265 if err != nil {
266 return *new(CacheT), err
267 }
268
269 cached = sc.makeCached(e, sc.entityUpdated)
270
271 sc.mu.Lock()
272 sc.cached[id] = cached
273 sc.lru.Add(id)
274 sc.mu.Unlock()
275
276 sc.evictIfNeeded()
277
278 return cached, nil
279}
280
281// ResolvePrefix retrieve an entity matching an id prefix. It fails if multiple
282// entity match.
283func (sc *SubCache[EntityT, ExcerptT, CacheT]) ResolvePrefix(prefix string) (CacheT, error) {
284 return sc.ResolveMatcher(func(excerpt ExcerptT) bool {
285 return excerpt.Id().HasPrefix(prefix)
286 })
287}
288
289func (sc *SubCache[EntityT, ExcerptT, CacheT]) ResolveMatcher(f func(ExcerptT) bool) (CacheT, error) {
290 id, err := sc.resolveMatcher(f)
291 if err != nil {
292 return *new(CacheT), err
293 }
294 return sc.Resolve(id)
295}
296
297// ResolveExcerpt retrieve an Excerpt matching the exact given id
298func (sc *SubCache[EntityT, ExcerptT, CacheT]) ResolveExcerpt(id entity.Id) (ExcerptT, error) {
299 sc.mu.RLock()
300 defer sc.mu.RUnlock()
301
302 excerpt, ok := sc.excerpts[id]
303 if !ok {
304 return *new(ExcerptT), entity.NewErrNotFound(sc.typename)
305 }
306
307 return excerpt, nil
308}
309
310// ResolveExcerptPrefix retrieve an Excerpt matching an id prefix. It fails if multiple
311// entity match.
312func (sc *SubCache[EntityT, ExcerptT, CacheT]) ResolveExcerptPrefix(prefix string) (ExcerptT, error) {
313 return sc.ResolveExcerptMatcher(func(excerpt ExcerptT) bool {
314 return excerpt.Id().HasPrefix(prefix)
315 })
316}
317
318func (sc *SubCache[EntityT, ExcerptT, CacheT]) ResolveExcerptMatcher(f func(ExcerptT) bool) (ExcerptT, error) {
319 id, err := sc.resolveMatcher(f)
320 if err != nil {
321 return *new(ExcerptT), err
322 }
323 return sc.ResolveExcerpt(id)
324}
325
326func (sc *SubCache[EntityT, ExcerptT, CacheT]) resolveMatcher(f func(ExcerptT) bool) (entity.Id, error) {
327 sc.mu.RLock()
328 defer sc.mu.RUnlock()
329
330 // preallocate but empty
331 matching := make([]entity.Id, 0, 5)
332
333 for _, excerpt := range sc.excerpts {
334 if f(excerpt) {
335 matching = append(matching, excerpt.Id())
336 }
337 }
338
339 if len(matching) > 1 {
340 return entity.UnsetId, entity.NewErrMultipleMatch(sc.typename, matching)
341 }
342
343 if len(matching) == 0 {
344 return entity.UnsetId, entity.NewErrNotFound(sc.typename)
345 }
346
347 return matching[0], nil
348}
349
350func (sc *SubCache[EntityT, ExcerptT, CacheT]) add(e EntityT) (CacheT, error) {
351 sc.mu.Lock()
352 if _, has := sc.cached[e.Id()]; has {
353 sc.mu.Unlock()
354 return *new(CacheT), fmt.Errorf("entity %s already exist in the cache", e.Id())
355 }
356
357 cached := sc.makeCached(e, sc.entityUpdated)
358 sc.cached[e.Id()] = cached
359 sc.lru.Add(e.Id())
360 sc.mu.Unlock()
361
362 sc.evictIfNeeded()
363
364 // force the write of the excerpt
365 err := sc.entityUpdated(e.Id())
366 if err != nil {
367 return *new(CacheT), err
368 }
369
370 return cached, nil
371}
372
373func (sc *SubCache[EntityT, ExcerptT, CacheT]) Remove(prefix string) error {
374 e, err := sc.ResolvePrefix(prefix)
375 if err != nil {
376 return err
377 }
378
379 sc.mu.Lock()
380
381 err = sc.actions.Remove(sc.repo, e.Id())
382 if err != nil {
383 sc.mu.Unlock()
384 return err
385 }
386
387 delete(sc.cached, e.Id())
388 delete(sc.excerpts, e.Id())
389 sc.lru.Remove(e.Id())
390
391 sc.mu.Unlock()
392
393 return sc.write()
394}
395
396func (sc *SubCache[EntityT, ExcerptT, CacheT]) MergeAll(remote string) <-chan entity.MergeResult {
397 out := make(chan entity.MergeResult)
398
399 // Intercept merge results to update the cache properly
400 go func() {
401 defer close(out)
402
403 author, err := sc.getUserIdentity()
404 if err != nil {
405 out <- entity.NewMergeError(err, "")
406 return
407 }
408
409 results := sc.actions.MergeAll(sc.repo, sc.resolvers(), remote, author)
410 for result := range results {
411 out <- result
412
413 if result.Err != nil {
414 continue
415 }
416
417 switch result.Status {
418 case entity.MergeStatusNew, entity.MergeStatusUpdated:
419 e := result.Entity.(EntityT)
420
421 // TODO: doesn't actually record in cache, should we?
422 cached := sc.makeCached(e, sc.entityUpdated)
423
424 sc.mu.Lock()
425 sc.excerpts[result.Id] = sc.makeExcerpt(cached)
426 sc.mu.Unlock()
427 }
428 }
429
430 err = sc.write()
431 if err != nil {
432 out <- entity.NewMergeError(err, "")
433 return
434 }
435 }()
436
437 return out
438
439}
440
441func (sc *SubCache[EntityT, ExcerptT, CacheT]) GetNamespace() string {
442 return sc.namespace
443}
444
445// entityUpdated is a callback to trigger when the excerpt of an entity changed
446func (sc *SubCache[EntityT, ExcerptT, CacheT]) entityUpdated(id entity.Id) error {
447 sc.mu.Lock()
448 e, ok := sc.cached[id]
449 if !ok {
450 sc.mu.Unlock()
451
452 // if the bug is not loaded at this point, it means it was loaded before
453 // but got evicted. Which means we potentially have multiple copies in
454 // memory and thus concurrent write.
455 // Failing immediately here is the simple and safe solution to avoid
456 // complicated data loss.
457 return errors.New("entity missing from cache")
458 }
459 sc.lru.Get(id)
460 // sc.excerpts[id] = bug2.NewBugExcerpt(b.bug, b.Snapshot())
461 sc.excerpts[id] = sc.makeExcerpt(e)
462 sc.mu.Unlock()
463
464 index, err := sc.repo.GetIndex(sc.typename)
465 if err != nil {
466 return err
467 }
468
469 err = index.IndexOne(e.Id().String(), sc.makeIndexData(e))
470 if err != nil {
471 return err
472 }
473
474 return sc.write()
475}
476
477// evictIfNeeded will evict an entity from the cache if needed
478func (sc *SubCache[EntityT, ExcerptT, CacheT]) evictIfNeeded() {
479 sc.mu.Lock()
480 defer sc.mu.Unlock()
481 if sc.lru.Len() <= sc.maxLoaded {
482 return
483 }
484
485 for _, id := range sc.lru.GetOldestToNewest() {
486 b := sc.cached[id]
487 if b.NeedCommit() {
488 continue
489 }
490
491 // TODO
492 // b.Lock()
493 sc.lru.Remove(id)
494 delete(sc.cached, id)
495
496 if sc.lru.Len() <= sc.maxLoaded {
497 return
498 }
499 }
500}