sanin.dev
oppi.li
+17
-5
appview/state/knotstream.go
+17
-5
appview/state/knotstream.go
···
13
13
kc "tangled.sh/tangled.sh/core/knotclient"
14
14
"tangled.sh/tangled.sh/core/log"
15
15
"tangled.sh/tangled.sh/core/rbac"
16
+
17
+
"github.com/posthog/posthog-go"
16
18
)
17
19
18
-
func KnotstreamConsumer(c *config.Config, d *db.DB, enforcer *rbac.Enforcer) (*kc.EventConsumer, error) {
20
+
func KnotstreamConsumer(c *config.Config, d *db.DB, enforcer *rbac.Enforcer, posthog posthog.Client) (*kc.EventConsumer, error) {
19
21
knots, err := db.GetCompletedRegistrations(d)
20
22
if err != nil {
21
23
return nil, err
···
33
35
34
36
cfg := kc.ConsumerConfig{
35
37
Sources: srcs,
36
-
ProcessFunc: knotstreamIngester(d, enforcer),
38
+
ProcessFunc: knotstreamIngester(d, enforcer, posthog, c.Core.Dev),
37
39
RetryInterval: c.Knotstream.RetryInterval,
38
40
MaxRetryInterval: c.Knotstream.MaxRetryInterval,
39
41
ConnectionTimeout: c.Knotstream.ConnectionTimeout,
···
47
49
return kc.NewEventConsumer(cfg), nil
48
50
}
49
51
50
-
func knotstreamIngester(d *db.DB, enforcer *rbac.Enforcer) kc.ProcessFunc {
52
+
func knotstreamIngester(d *db.DB, enforcer *rbac.Enforcer, posthog posthog.Client, dev bool) kc.ProcessFunc {
51
53
return func(source kc.EventSource, msg kc.Message) error {
52
54
switch msg.Nsid {
53
55
case tangled.GitRefUpdateNSID:
54
-
return ingestRefUpdate(d, enforcer, source, msg)
56
+
return ingestRefUpdate(d, enforcer, posthog, dev, source, msg)
55
57
case tangled.PipelineNSID:
56
58
// TODO
57
59
}
···
60
62
}
61
63
}
62
64
63
-
func ingestRefUpdate(d *db.DB, enforcer *rbac.Enforcer, source kc.EventSource, msg kc.Message) error {
65
+
func ingestRefUpdate(d *db.DB, enforcer *rbac.Enforcer, pc posthog.Client, dev bool, source kc.EventSource, msg kc.Message) error {
64
66
var record tangled.GitRefUpdate
65
67
err := json.Unmarshal(msg.EventJson, &record)
66
68
if err != nil {
···
104
106
}
105
107
if err := db.AddPunch(d, punch); err != nil {
106
108
return err
109
+
}
110
+
111
+
if !dev {
112
+
err = pc.Enqueue(posthog.Capture{
113
+
DistinctId: record.CommitterDid,
114
+
Event: "git_ref_update",
115
+
})
116
+
if err != nil {
117
+
// non-fatal, TODO: log this
118
+
}
107
119
}
108
120
109
121
return nil
+1
-1
appview/state/state.go
+1
-1
appview/state/state.go
···
109
109
return nil, fmt.Errorf("failed to start jetstream watcher: %w", err)
110
110
}
111
111
112
-
knotstream, err := KnotstreamConsumer(config, d, enforcer)
112
+
knotstream, err := KnotstreamConsumer(config, d, enforcer, posthog)
113
113
if err != nil {
114
114
return nil, fmt.Errorf("failed to start knotstream consumer: %w", err)
115
115
}