1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
package market
import (
"context"
"time"
"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/ttlcache"
)
const HeartbeatEventDebounce = 15 * time.Second
type heartbeatEntry struct {
at time.Time
checkedAt time.Time
valid bool
}
type HeartbeatCache struct {
Getter RecordGetter
held *ttlcache.Cache[string, heartbeatEntry]
}
func NewHeartbeatCache(getter RecordGetter) *HeartbeatCache {
return &HeartbeatCache{
Getter: getter,
held: ttlcache.New[string, heartbeatEntry](ttlcache.Options{TTL: HeartbeatTTL + HeartbeatMaxSkew}),
}
}
func heartbeatFreshAt(entry heartbeatEntry, now time.Time) bool {
return entry.valid && now.Sub(entry.at) <= HeartbeatTTL
}
func (c *HeartbeatCache) read(ctx context.Context, did string, now time.Time) heartbeatEntry {
entry := heartbeatEntry{checkedAt: now}
if c.Getter == nil {
return entry
}
rec, err := c.Getter.GetRecord(ctx, did, HeartbeatNSID, HeartbeatRKey)
if err != nil {
return entry
}
value, _ := rec.Value["createdAt"].(string)
at, err := ParseTime(value)
if err != nil || at.After(now.Add(HeartbeatMaxSkew)) {
return entry
}
entry.at = at
entry.valid = true
return entry
}
func (c *HeartbeatCache) FreshDIDs(ctx context.Context, dids []string, now time.Time) []string {
out := make([]string, 0, len(dids))
for _, did := range dids {
entry, ok := c.held.Get(did)
if !ok || !(heartbeatFreshAt(entry, now) || now.Sub(entry.checkedAt) < HeartbeatInterval) {
entry = c.read(ctx, did, now)
c.held.Set(did, entry)
}
if heartbeatFreshAt(entry, now) {
out = append(out, did)
}
}
return out
}
func (c *HeartbeatCache) HandleEvent(ctx context.Context, event WatchEvent) {
if event.Collection != HeartbeatNSID || event.DID == "" {
return
}
now := time.Now()
if event.Operation == "delete" {
c.held.Set(event.DID, heartbeatEntry{checkedAt: now})
return
}
held, ok := c.held.Get(event.DID)
if ok && now.Sub(held.checkedAt) < HeartbeatEventDebounce {
return
}
entry := c.read(ctx, event.DID, now)
c.held.Set(event.DID, entry)
}
