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