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