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
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
package marketpresence
import (
"context"
"crypto/sha256"
"encoding/hex"
"strings"
"sync"
"time"
"github.com/bluesky-social/indigo/atproto/identity"
"github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/ingress"
"github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/market"
)
const (
MaxPresenceName = 253
ReasonFresh = "HeartbeatFresh"
ReasonNoDID = "NoDID"
ReasonNoHeartbeat = "NoHeartbeat"
ReasonInvalid = "HeartbeatInvalid"
ReasonMismatch = "HeartbeatMismatch"
ReasonInFuture = "HeartbeatInFuture"
)
func PresenceName(bidder string) string {
name := "bidder-" + bidder
if len(name) <= MaxPresenceName {
return name
}
sum := sha256.Sum256([]byte(bidder))
return "bidder-" + hex.EncodeToString(sum[:])[:32]
}
type Config struct {
Reader market.RecordGetter
Directory identity.Directory
HTTPClient market.HTTPDoer
}
type Query struct {
DID string
OfferingURI string
PLCDirectoryURL string
Now time.Time
}
type Observation struct {
At time.Time
Reason string
}
type Heartbeats struct {
reader market.RecordGetter
directory identity.Directory
http market.HTTPDoer
mu sync.Mutex
readers map[string]market.RecordGetter
}
func New(cfg Config) *Heartbeats {
return &Heartbeats{reader: cfg.Reader, directory: cfg.Directory, http: cfg.HTTPClient, readers: map[string]market.RecordGetter{}}
}
func (h *Heartbeats) Observe(ctx context.Context, q Query) Observation {
did := strings.TrimSpace(q.DID)
if did == "" {
return Observation{Reason: ReasonNoDID}
}
rec, err := h.readerFor(strings.TrimSpace(q.PLCDirectoryURL)).GetRecord(ctx, did, market.HeartbeatNSID, market.HeartbeatRKey)
if err != nil {
return Observation{Reason: ReasonNoHeartbeat}
}
heartbeat, err := market.ParseHeartbeat(rec.Value)
if err != nil {
return Observation{Reason: ReasonInvalid}
}
if offering := strings.TrimSpace(q.OfferingURI); offering != "" && heartbeat.Offering != offering {
return Observation{Reason: ReasonMismatch}
}
at, err := market.ParseTime(heartbeat.CreatedAt)
if err != nil {
return Observation{Reason: ReasonInvalid}
}
if at.After(q.Now.Add(market.HeartbeatMaxSkew)) {
return Observation{Reason: ReasonInFuture}
}
return Observation{At: at, Reason: ReasonFresh}
}
func (h *Heartbeats) readerFor(plc string) market.RecordGetter {
h.mu.Lock()
defer h.mu.Unlock()
if h.reader != nil {
return h.reader
}
if reader, ok := h.readers[plc]; ok {
return reader
}
dir := h.directory
if dir == nil {
dir = ingress.NewDirectory(plc)
}
reader := market.NewHTTPReader(dir)
if h.http != nil {
reader.Client = h.http
}
h.readers[plc] = reader
return reader
}
