tangled.org
Trigger webhooks on git push events from knotstream.
Signed-off-by: Anirudh Oppiliappan anirudh@tangled.org
+30
-13
appview/state/knotstream.go
+30
-13
appview/state/knotstream.go
···
5
5
"encoding/json"
6
6
"errors"
7
7
"fmt"
8
-
"slices"
9
8
"time"
10
9
10
+
"tangled.org/core/appview/notify"
11
+
11
12
"tangled.org/core/api/tangled"
12
13
"tangled.org/core/appview/cache"
13
14
"tangled.org/core/appview/config"
···
25
26
"github.com/posthog/posthog-go"
26
27
)
27
28
28
-
func Knotstream(ctx context.Context, c *config.Config, d *db.DB, enforcer *rbac.Enforcer, posthog posthog.Client) (*ec.Consumer, error) {
29
+
func Knotstream(ctx context.Context, c *config.Config, d *db.DB, enforcer *rbac.Enforcer, posthog posthog.Client, notifier notify.Notifier) (*ec.Consumer, error) {
29
30
logger := log.FromContext(ctx)
30
31
logger = log.SubLogger(logger, "knotstream")
31
32
···
48
49
49
50
cfg := ec.ConsumerConfig{
50
51
Sources: srcs,
51
-
ProcessFunc: knotIngester(d, enforcer, posthog, c.Core.Dev),
52
+
ProcessFunc: knotIngester(d, enforcer, posthog, notifier, c.Core.Dev),
52
53
RetryInterval: c.Knotstream.RetryInterval,
53
54
MaxRetryInterval: c.Knotstream.MaxRetryInterval,
54
55
ConnectionTimeout: c.Knotstream.ConnectionTimeout,
···
62
63
return ec.NewConsumer(cfg), nil
63
64
}
64
65
65
-
func knotIngester(d *db.DB, enforcer *rbac.Enforcer, posthog posthog.Client, dev bool) ec.ProcessFunc {
66
+
func knotIngester(d *db.DB, enforcer *rbac.Enforcer, posthog posthog.Client, notifier notify.Notifier, dev bool) ec.ProcessFunc {
66
67
return func(ctx context.Context, source ec.Source, msg ec.Message) error {
67
68
switch msg.Nsid {
68
69
case tangled.GitRefUpdateNSID:
69
-
return ingestRefUpdate(d, enforcer, posthog, dev, source, msg)
70
+
return ingestRefUpdate(d, enforcer, posthog, notifier, dev, source, msg, ctx)
70
71
case tangled.PipelineNSID:
71
72
return ingestPipeline(d, source, msg)
72
73
}
···
75
76
}
76
77
}
77
78
78
-
func ingestRefUpdate(d *db.DB, enforcer *rbac.Enforcer, pc posthog.Client, dev bool, source ec.Source, msg ec.Message) error {
79
+
func ingestRefUpdate(d *db.DB, enforcer *rbac.Enforcer, pc posthog.Client, notifier notify.Notifier, dev bool, source ec.Source, msg ec.Message, ctx context.Context) error {
80
+
logger := log.FromContext(ctx)
81
+
79
82
var record tangled.GitRefUpdate
80
83
err := json.Unmarshal(msg.EventJson, &record)
81
84
if err != nil {
82
85
return err
83
86
}
84
87
85
-
knownKnots, err := enforcer.GetKnotsForUser(record.CommitterDid)
86
-
if err != nil {
87
-
return err
88
-
}
89
-
if !slices.Contains(knownKnots, source.Key()) {
90
-
return fmt.Errorf("%s does not belong to %s, something is fishy", record.CommitterDid, source.Key())
88
+
logger.Info("processing gitRefUpdate event",
89
+
"repo_did", record.RepoDid,
90
+
"repo_name", record.RepoName,
91
+
"ref", record.Ref,
92
+
"old_sha", record.OldSha,
93
+
"new_sha", record.NewSha)
94
+
95
+
// trigger webhook notifications first (before other ops that might fail)
96
+
repos, err := db.GetRepos(
97
+
d,
98
+
0,
99
+
orm.FilterEq("did", record.RepoDid),
100
+
orm.FilterEq("name", record.RepoName),
101
+
)
102
+
if err == nil && len(repos) == 1 {
103
+
notifier.Push(ctx, &repos[0], record.Ref, record.OldSha, record.NewSha, record.CommitterDid)
91
104
}
92
105
93
106
err1 := populatePunchcard(d, record)
94
107
err2 := updateRepoLanguages(d, record)
95
108
96
109
var err3 error
97
-
if !dev {
110
+
if !dev && record.CommitterDid != "" {
98
111
err3 = pc.Enqueue(posthog.Capture{
99
112
DistinctId: record.CommitterDid,
100
113
Event: "git_ref_update",
···
105
118
}
106
119
107
120
func populatePunchcard(d *db.DB, record tangled.GitRefUpdate) error {
121
+
if record.CommitterDid == "" {
122
+
return nil
123
+
}
124
+
108
125
knownEmails, err := db.GetAllEmails(d, record.CommitterDid)
109
126
if err != nil {
110
127
return err
+16
-12
appview/state/state.go
+16
-12
appview/state/state.go
···
151
151
return nil, fmt.Errorf("failed to start jetstream watcher: %w", err)
152
152
}
153
153
154
-
knotstream, err := Knotstream(ctx, config, d, enforcer, posthog)
155
-
if err != nil {
156
-
return nil, fmt.Errorf("failed to start knotstream consumer: %w", err)
157
-
}
158
-
knotstream.Start(ctx)
159
-
160
-
spindlestream, err := Spindlestream(ctx, config, d, enforcer)
161
-
if err != nil {
162
-
return nil, fmt.Errorf("failed to start spindlestream consumer: %w", err)
163
-
}
164
-
spindlestream.Start(ctx)
165
-
166
154
var notifiers []notify.Notifier
167
155
168
156
// Always add the database notifier
···
173
161
notifiers = append(notifiers, phnotify.NewPosthogNotifier(posthog))
174
162
}
175
163
notifiers = append(notifiers, indexer)
164
+
165
+
// Add webhook notifier
166
+
notifiers = append(notifiers, notify.NewWebhookNotifier(d))
167
+
176
168
notifier := notify.NewMergedNotifier(notifiers)
177
169
notifier = notify.NewLoggingNotifier(notifier, tlog.SubLogger(logger, "notify"))
178
170
171
+
knotstream, err := Knotstream(ctx, config, d, enforcer, posthog, notifier)
172
+
if err != nil {
173
+
return nil, fmt.Errorf("failed to start knotstream consumer: %w", err)
174
+
}
175
+
knotstream.Start(ctx)
176
+
177
+
spindlestream, err := Spindlestream(ctx, config, d, enforcer)
178
+
if err != nil {
179
+
return nil, fmt.Errorf("failed to start spindlestream consumer: %w", err)
180
+
}
181
+
spindlestream.Start(ctx)
182
+
179
183
state := &State{
180
184
d,
181
185
notifier,
History
6 rounds
1 comment
anirudh.fi
submitted
#5
1 commit
expand
collapse
appview/state: integrate webhooks with gitRefUpdate events
Trigger webhooks on git push events from knotstream.
Signed-off-by: Anirudh Oppiliappan <anirudh@tangled.org>
3/3 success
expand
collapse
expand 0 comments
pull request successfully merged
anirudh.fi
submitted
#4
1 commit
expand
collapse
appview/state: integrate webhooks with gitRefUpdate events
Trigger webhooks on git push events from knotstream.
Signed-off-by: Anirudh Oppiliappan <anirudh@tangled.org>
3/3 success
expand
collapse
expand 0 comments
anirudh.fi
submitted
#3
1 commit
expand
collapse
appview/state: integrate webhooks with gitRefUpdate events
Trigger webhooks on git push events from knotstream.
Signed-off-by: Anirudh Oppiliappan <anirudh@tangled.org>
3/3 success
expand
collapse
expand 0 comments
anirudh.fi
submitted
#2
1 commit
expand
collapse
appview/state: integrate webhooks with gitRefUpdate events
Trigger webhooks on git push events from knotstream.
Signed-off-by: Anirudh Oppiliappan <anirudh@tangled.org>
3/3 success
expand
collapse
expand 0 comments
anirudh.fi
submitted
#1
1 commit
expand
collapse
appview/state: integrate webhooks with gitRefUpdate events
Trigger webhooks on git push events from knotstream.
Signed-off-by: Anirudh Oppiliappan <anirudh@tangled.org>
3/3 success
expand
collapse
expand 1 comment
anirudh.fi
submitted
#0
1 commit
expand
collapse
appview/state: integrate webhooks with gitRefUpdate events
Trigger webhooks on git push events from knotstream.
Signed-off-by: Anirudh Oppiliappan <anirudh@tangled.org>
Joined, we can do the same here (collect the error and join in the final statementchangeset lgtm otherwise!