Monorepo for Tangled
appview/notify: move logging and webhook notifiers to their own packages
Signed-off-by: Anirudh Oppiliappan <anirudh@tangled.org>
authored by
Anirudh Oppiliappan
and committed by
tangled.org
(Apr 2, 2026, 7:37 PM UTC)
13ec3ea8
88b2cbf2
+350
-369
-125
appview/notify/logging_notifier.go
-125
appview/notify/logging_notifier.go
···
1
-
package notify
2
-
3
-
import (
4
-
"context"
5
-
"log/slog"
6
-
7
-
"tangled.org/core/appview/models"
8
-
tlog "tangled.org/core/log"
9
-
10
-
"github.com/bluesky-social/indigo/atproto/syntax"
11
-
)
12
-
13
-
type loggingNotifier struct {
14
-
inner Notifier
15
-
logger *slog.Logger
16
-
}
17
-
18
-
func NewLoggingNotifier(inner Notifier, logger *slog.Logger) Notifier {
19
-
return &loggingNotifier{
20
-
inner,
21
-
logger,
22
-
}
23
-
}
24
-
25
-
var _ Notifier = &loggingNotifier{}
26
-
27
-
func (l *loggingNotifier) NewRepo(ctx context.Context, repo *models.Repo) {
28
-
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "NewRepo"))
29
-
l.inner.NewRepo(ctx, repo)
30
-
}
31
-
32
-
func (l *loggingNotifier) NewStar(ctx context.Context, star *models.Star) {
33
-
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "NewStar"))
34
-
l.inner.NewStar(ctx, star)
35
-
}
36
-
37
-
func (l *loggingNotifier) DeleteStar(ctx context.Context, star *models.Star) {
38
-
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "DeleteStar"))
39
-
l.inner.DeleteStar(ctx, star)
40
-
}
41
-
42
-
func (l *loggingNotifier) NewIssue(ctx context.Context, issue *models.Issue, mentions []syntax.DID) {
43
-
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "NewIssue"))
44
-
l.inner.NewIssue(ctx, issue, mentions)
45
-
}
46
-
47
-
func (l *loggingNotifier) NewIssueComment(ctx context.Context, comment *models.IssueComment, mentions []syntax.DID) {
48
-
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "NewIssueComment"))
49
-
l.inner.NewIssueComment(ctx, comment, mentions)
50
-
}
51
-
52
-
func (l *loggingNotifier) NewIssueState(ctx context.Context, actor syntax.DID, issue *models.Issue) {
53
-
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "NewIssueState"))
54
-
l.inner.NewIssueState(ctx, actor, issue)
55
-
}
56
-
57
-
func (l *loggingNotifier) DeleteIssue(ctx context.Context, issue *models.Issue) {
58
-
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "DeleteIssue"))
59
-
l.inner.DeleteIssue(ctx, issue)
60
-
}
61
-
62
-
func (l *loggingNotifier) NewIssueLabelOp(ctx context.Context, issue *models.Issue) {
63
-
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "NewIssueLabelOp"))
64
-
l.inner.NewIssueLabelOp(ctx, issue)
65
-
}
66
-
67
-
func (l *loggingNotifier) NewPullLabelOp(ctx context.Context, pull *models.Pull) {
68
-
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "NewPullLabelOp"))
69
-
l.inner.NewPullLabelOp(ctx, pull)
70
-
}
71
-
72
-
func (l *loggingNotifier) NewFollow(ctx context.Context, follow *models.Follow) {
73
-
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "NewFollow"))
74
-
l.inner.NewFollow(ctx, follow)
75
-
}
76
-
77
-
func (l *loggingNotifier) DeleteFollow(ctx context.Context, follow *models.Follow) {
78
-
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "DeleteFollow"))
79
-
l.inner.DeleteFollow(ctx, follow)
80
-
}
81
-
82
-
func (l *loggingNotifier) NewPull(ctx context.Context, pull *models.Pull) {
83
-
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "NewPull"))
84
-
l.inner.NewPull(ctx, pull)
85
-
}
86
-
87
-
func (l *loggingNotifier) NewPullComment(ctx context.Context, comment *models.PullComment, mentions []syntax.DID) {
88
-
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "NewPullComment"))
89
-
l.inner.NewPullComment(ctx, comment, mentions)
90
-
}
91
-
92
-
func (l *loggingNotifier) NewPullState(ctx context.Context, actor syntax.DID, pull *models.Pull) {
93
-
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "NewPullState"))
94
-
l.inner.NewPullState(ctx, actor, pull)
95
-
}
96
-
97
-
func (l *loggingNotifier) UpdateProfile(ctx context.Context, profile *models.Profile) {
98
-
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "UpdateProfile"))
99
-
l.inner.UpdateProfile(ctx, profile)
100
-
}
101
-
102
-
func (l *loggingNotifier) NewString(ctx context.Context, s *models.String) {
103
-
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "NewString"))
104
-
l.inner.NewString(ctx, s)
105
-
}
106
-
107
-
func (l *loggingNotifier) EditString(ctx context.Context, s *models.String) {
108
-
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "EditString"))
109
-
l.inner.EditString(ctx, s)
110
-
}
111
-
112
-
func (l *loggingNotifier) DeleteString(ctx context.Context, did, rkey string) {
113
-
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "DeleteString"))
114
-
l.inner.DeleteString(ctx, did, rkey)
115
-
}
116
-
117
-
func (l *loggingNotifier) Push(ctx context.Context, repo *models.Repo, ref, oldSha, newSha, committerDid string) {
118
-
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "Push"))
119
-
l.inner.Push(ctx, repo, ref, oldSha, newSha, committerDid)
120
-
}
121
-
122
-
func (l *loggingNotifier) Clone(ctx context.Context, repo *models.Repo) {
123
-
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "Clone"))
124
-
l.inner.Clone(ctx, repo)
125
-
}
-241
appview/notify/webhook_notifier.go
-241
appview/notify/webhook_notifier.go
···
1
-
package notify
2
-
3
-
import (
4
-
"bytes"
5
-
"context"
6
-
"crypto/hmac"
7
-
"crypto/sha256"
8
-
"encoding/hex"
9
-
"encoding/json"
10
-
"fmt"
11
-
"io"
12
-
"log/slog"
13
-
"net/http"
14
-
"time"
15
-
16
-
"github.com/avast/retry-go/v4"
17
-
"github.com/google/uuid"
18
-
"tangled.org/core/appview/db"
19
-
"tangled.org/core/appview/models"
20
-
"tangled.org/core/log"
21
-
)
22
-
23
-
type WebhookNotifier struct {
24
-
BaseNotifier
25
-
db *db.DB
26
-
logger *slog.Logger
27
-
client *http.Client
28
-
}
29
-
30
-
func NewWebhookNotifier(database *db.DB) *WebhookNotifier {
31
-
return &WebhookNotifier{
32
-
db: database,
33
-
logger: log.New("webhook-notifier"),
34
-
client: &http.Client{
35
-
Timeout: 30 * time.Second,
36
-
},
37
-
}
38
-
}
39
-
40
-
// Push implements the Notifier interface for git push events
41
-
func (w *WebhookNotifier) Push(ctx context.Context, repo *models.Repo, ref, oldSha, newSha, committerDid string) {
42
-
webhooks, err := db.GetActiveWebhooksForRepo(w.db, repo.RepoAt())
43
-
if err != nil {
44
-
w.logger.Error("failed to get webhooks for repo", "repo", repo.RepoAt(), "err", err)
45
-
return
46
-
}
47
-
48
-
// check if any webhooks are subscribed to push events
49
-
var pushWebhooks []models.Webhook
50
-
for _, webhook := range webhooks {
51
-
if webhook.HasEvent(models.WebhookEventPush) {
52
-
pushWebhooks = append(pushWebhooks, webhook)
53
-
}
54
-
}
55
-
56
-
if len(pushWebhooks) == 0 {
57
-
return
58
-
}
59
-
60
-
payload, err := w.buildPushPayload(repo, ref, oldSha, newSha, committerDid)
61
-
if err != nil {
62
-
w.logger.Error("failed to build push payload", "repo", repo.RepoAt(), "err", err)
63
-
return
64
-
}
65
-
66
-
// Send webhooks
67
-
for _, webhook := range pushWebhooks {
68
-
go w.sendWebhook(ctx, webhook, string(models.WebhookEventPush), payload)
69
-
}
70
-
}
71
-
72
-
func (w *WebhookNotifier) Clone(ctx context.Context, repo *models.Repo) {}
73
-
74
-
// buildPushPayload creates the webhook payload
75
-
func (w *WebhookNotifier) buildPushPayload(repo *models.Repo, ref, oldSha, newSha, committerDid string) (*models.WebhookPayload, error) {
76
-
owner := repo.Did
77
-
78
-
pusher := committerDid
79
-
if committerDid == "" {
80
-
pusher = owner
81
-
}
82
-
83
-
// Build repository object
84
-
repository := models.WebhookRepository{
85
-
Name: repo.Name,
86
-
FullName: fmt.Sprintf("%s/%s", repo.Did, repo.Name),
87
-
Description: repo.Description,
88
-
Fork: repo.Source != "",
89
-
HtmlUrl: fmt.Sprintf("https://%s/%s/%s", repo.Knot, repo.Did, repo.Name),
90
-
CloneUrl: fmt.Sprintf("https://%s/%s/%s", repo.Knot, repo.Did, repo.Name),
91
-
SshUrl: fmt.Sprintf("ssh://git@%s/%s/%s", repo.Knot, repo.Did, repo.Name),
92
-
CreatedAt: repo.Created.Format(time.RFC3339),
93
-
UpdatedAt: repo.Created.Format(time.RFC3339),
94
-
Owner: models.WebhookUser{
95
-
Did: owner,
96
-
},
97
-
}
98
-
99
-
// Add optional fields
100
-
if repo.Website != "" {
101
-
repository.Website = repo.Website
102
-
}
103
-
if repo.RepoStats != nil {
104
-
repository.StarsCount = repo.RepoStats.StarCount
105
-
repository.OpenIssues = repo.RepoStats.IssueCount.Open
106
-
}
107
-
108
-
// Build payload
109
-
payload := &models.WebhookPayload{
110
-
Ref: ref,
111
-
Before: oldSha,
112
-
After: newSha,
113
-
Repository: repository,
114
-
Pusher: models.WebhookUser{
115
-
Did: pusher,
116
-
},
117
-
}
118
-
119
-
return payload, nil
120
-
}
121
-
122
-
// sendWebhook sends the webhook http request
123
-
func (w *WebhookNotifier) sendWebhook(ctx context.Context, webhook models.Webhook, event string, payload *models.WebhookPayload) {
124
-
deliveryId := uuid.New().String()
125
-
126
-
payloadBytes, err := json.Marshal(payload)
127
-
if err != nil {
128
-
w.logger.Error("failed to marshal webhook payload", "webhook_id", webhook.Id, "err", err)
129
-
return
130
-
}
131
-
132
-
req, err := http.NewRequestWithContext(ctx, "POST", webhook.Url, bytes.NewReader(payloadBytes))
133
-
if err != nil {
134
-
w.logger.Error("failed to create webhook request", "webhook_id", webhook.Id, "err", err)
135
-
return
136
-
}
137
-
138
-
shortSha := payload.After[:7]
139
-
140
-
req.Header.Set("Content-Type", "application/json")
141
-
req.Header.Set("User-Agent", "Tangled-Hook/"+shortSha)
142
-
req.Header.Set("X-Tangled-Event", event)
143
-
req.Header.Set("X-Tangled-Hook-ID", fmt.Sprintf("%d", webhook.Id))
144
-
req.Header.Set("X-Tangled-Delivery", deliveryId)
145
-
req.Header.Set("X-Tangled-Repo", payload.Repository.FullName)
146
-
147
-
if webhook.Secret != "" {
148
-
signature := w.computeSignature(payloadBytes, webhook.Secret)
149
-
req.Header.Set("X-Tangled-Signature-256", "sha256="+signature)
150
-
}
151
-
152
-
delivery := &models.WebhookDelivery{
153
-
WebhookId: webhook.Id,
154
-
Event: event,
155
-
DeliveryId: deliveryId,
156
-
Url: webhook.Url,
157
-
RequestBody: string(payloadBytes),
158
-
}
159
-
160
-
// retry webhook delivery with exponential backoff
161
-
retryOpts := []retry.Option{
162
-
retry.Attempts(3),
163
-
retry.Delay(1 * time.Second),
164
-
retry.MaxDelay(10 * time.Second),
165
-
retry.DelayType(retry.BackOffDelay),
166
-
retry.LastErrorOnly(true),
167
-
retry.OnRetry(func(n uint, err error) {
168
-
w.logger.Info("retrying webhook delivery",
169
-
"webhook_id", webhook.Id,
170
-
"attempt", n+1,
171
-
"err", err)
172
-
}),
173
-
retry.Context(ctx),
174
-
retry.RetryIf(func(err error) bool {
175
-
// only retry on network errors or 5xx responses
176
-
if err != nil {
177
-
return true
178
-
}
179
-
return false
180
-
}),
181
-
}
182
-
183
-
var resp *http.Response
184
-
err = retry.Do(func() error {
185
-
var err error
186
-
resp, err = w.client.Do(req)
187
-
if err != nil {
188
-
return err
189
-
}
190
-
191
-
// retry on 5xx server errors
192
-
if resp.StatusCode >= 500 {
193
-
defer resp.Body.Close()
194
-
return fmt.Errorf("server error: %d", resp.StatusCode)
195
-
}
196
-
197
-
return nil
198
-
}, retryOpts...)
199
-
200
-
if err != nil {
201
-
w.logger.Error("webhook request failed after retries", "webhook_id", webhook.Id, "err", err)
202
-
delivery.Success = false
203
-
delivery.ResponseBody = err.Error()
204
-
} else {
205
-
defer resp.Body.Close()
206
-
207
-
delivery.ResponseCode = resp.StatusCode
208
-
delivery.Success = resp.StatusCode >= 200 && resp.StatusCode < 300
209
-
210
-
// Read response body (limit to 10KB)
211
-
bodyBytes, err := io.ReadAll(io.LimitReader(resp.Body, 10*1024))
212
-
if err != nil {
213
-
w.logger.Warn("failed to read webhook response body", "webhook_id", webhook.Id, "err", err)
214
-
} else {
215
-
delivery.ResponseBody = string(bodyBytes)
216
-
}
217
-
218
-
if !delivery.Success {
219
-
w.logger.Warn("webhook delivery failed",
220
-
"webhook_id", webhook.Id,
221
-
"status", resp.StatusCode,
222
-
"url", webhook.Url)
223
-
} else {
224
-
w.logger.Info("webhook delivered successfully",
225
-
"webhook_id", webhook.Id,
226
-
"url", webhook.Url,
227
-
"delivery_id", deliveryId)
228
-
}
229
-
}
230
-
231
-
if err := db.AddWebhookDelivery(w.db, delivery); err != nil {
232
-
w.logger.Error("failed to record webhook delivery", "webhook_id", webhook.Id, "err", err)
233
-
}
234
-
}
235
-
236
-
// computeSignature computes HMAC-SHA256 signature for the payload
237
-
func (w *WebhookNotifier) computeSignature(payload []byte, secret string) string {
238
-
mac := hmac.New(sha256.New, []byte(secret))
239
-
mac.Write(payload)
240
-
return hex.EncodeToString(mac.Sum(nil))
241
-
}
+4
-3
appview/state/state.go
+4
-3
appview/state/state.go
···
21
21
"tangled.org/core/appview/models"
22
22
"tangled.org/core/appview/notify"
23
23
dbnotify "tangled.org/core/appview/notify/db"
24
+
lognotify "tangled.org/core/appview/notify/logging"
24
25
phnotify "tangled.org/core/appview/notify/posthog"
26
+
whnotify "tangled.org/core/appview/notify/webhook"
25
27
"tangled.org/core/appview/oauth"
26
28
"tangled.org/core/appview/pages"
27
29
"tangled.org/core/appview/reporesolver"
···
168
170
}
169
171
notifiers = append(notifiers, indexer)
170
172
171
-
// Add webhook notifier
172
-
notifiers = append(notifiers, notify.NewWebhookNotifier(d))
173
+
notifiers = append(notifiers, whnotify.NewNotifier(d))
173
174
174
175
notifier := notify.NewMergedNotifier(notifiers)
175
-
notifier = notify.NewLoggingNotifier(notifier, tlog.SubLogger(logger, "notify"))
176
+
notifier = lognotify.NewLoggingNotifier(notifier, tlog.SubLogger(logger, "notify"))
176
177
177
178
var cfClient *cloudflare.Client
178
179
if config.Cloudflare.ApiToken != "" {
+122
appview/notify/logging/notifier.go
+122
appview/notify/logging/notifier.go
···
1
+
package logging
2
+
3
+
import (
4
+
"context"
5
+
"log/slog"
6
+
7
+
"github.com/bluesky-social/indigo/atproto/syntax"
8
+
"tangled.org/core/appview/models"
9
+
"tangled.org/core/appview/notify"
10
+
tlog "tangled.org/core/log"
11
+
)
12
+
13
+
type loggingNotifier struct {
14
+
inner notify.Notifier
15
+
logger *slog.Logger
16
+
}
17
+
18
+
func NewLoggingNotifier(inner notify.Notifier, logger *slog.Logger) notify.Notifier {
19
+
return &loggingNotifier{inner, logger}
20
+
}
21
+
22
+
var _ notify.Notifier = &loggingNotifier{}
23
+
24
+
func (l *loggingNotifier) NewRepo(ctx context.Context, repo *models.Repo) {
25
+
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "NewRepo"))
26
+
l.inner.NewRepo(ctx, repo)
27
+
}
28
+
29
+
func (l *loggingNotifier) NewStar(ctx context.Context, star *models.Star) {
30
+
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "NewStar"))
31
+
l.inner.NewStar(ctx, star)
32
+
}
33
+
34
+
func (l *loggingNotifier) DeleteStar(ctx context.Context, star *models.Star) {
35
+
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "DeleteStar"))
36
+
l.inner.DeleteStar(ctx, star)
37
+
}
38
+
39
+
func (l *loggingNotifier) NewIssue(ctx context.Context, issue *models.Issue, mentions []syntax.DID) {
40
+
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "NewIssue"))
41
+
l.inner.NewIssue(ctx, issue, mentions)
42
+
}
43
+
44
+
func (l *loggingNotifier) NewIssueComment(ctx context.Context, comment *models.IssueComment, mentions []syntax.DID) {
45
+
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "NewIssueComment"))
46
+
l.inner.NewIssueComment(ctx, comment, mentions)
47
+
}
48
+
49
+
func (l *loggingNotifier) NewIssueState(ctx context.Context, actor syntax.DID, issue *models.Issue) {
50
+
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "NewIssueState"))
51
+
l.inner.NewIssueState(ctx, actor, issue)
52
+
}
53
+
54
+
func (l *loggingNotifier) DeleteIssue(ctx context.Context, issue *models.Issue) {
55
+
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "DeleteIssue"))
56
+
l.inner.DeleteIssue(ctx, issue)
57
+
}
58
+
59
+
func (l *loggingNotifier) NewIssueLabelOp(ctx context.Context, issue *models.Issue) {
60
+
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "NewIssueLabelOp"))
61
+
l.inner.NewIssueLabelOp(ctx, issue)
62
+
}
63
+
64
+
func (l *loggingNotifier) NewPullLabelOp(ctx context.Context, pull *models.Pull) {
65
+
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "NewPullLabelOp"))
66
+
l.inner.NewPullLabelOp(ctx, pull)
67
+
}
68
+
69
+
func (l *loggingNotifier) NewFollow(ctx context.Context, follow *models.Follow) {
70
+
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "NewFollow"))
71
+
l.inner.NewFollow(ctx, follow)
72
+
}
73
+
74
+
func (l *loggingNotifier) DeleteFollow(ctx context.Context, follow *models.Follow) {
75
+
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "DeleteFollow"))
76
+
l.inner.DeleteFollow(ctx, follow)
77
+
}
78
+
79
+
func (l *loggingNotifier) NewPull(ctx context.Context, pull *models.Pull) {
80
+
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "NewPull"))
81
+
l.inner.NewPull(ctx, pull)
82
+
}
83
+
84
+
func (l *loggingNotifier) NewPullComment(ctx context.Context, comment *models.PullComment, mentions []syntax.DID) {
85
+
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "NewPullComment"))
86
+
l.inner.NewPullComment(ctx, comment, mentions)
87
+
}
88
+
89
+
func (l *loggingNotifier) NewPullState(ctx context.Context, actor syntax.DID, pull *models.Pull) {
90
+
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "NewPullState"))
91
+
l.inner.NewPullState(ctx, actor, pull)
92
+
}
93
+
94
+
func (l *loggingNotifier) UpdateProfile(ctx context.Context, profile *models.Profile) {
95
+
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "UpdateProfile"))
96
+
l.inner.UpdateProfile(ctx, profile)
97
+
}
98
+
99
+
func (l *loggingNotifier) NewString(ctx context.Context, s *models.String) {
100
+
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "NewString"))
101
+
l.inner.NewString(ctx, s)
102
+
}
103
+
104
+
func (l *loggingNotifier) EditString(ctx context.Context, s *models.String) {
105
+
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "EditString"))
106
+
l.inner.EditString(ctx, s)
107
+
}
108
+
109
+
func (l *loggingNotifier) DeleteString(ctx context.Context, did, rkey string) {
110
+
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "DeleteString"))
111
+
l.inner.DeleteString(ctx, did, rkey)
112
+
}
113
+
114
+
func (l *loggingNotifier) Push(ctx context.Context, repo *models.Repo, ref, oldSha, newSha, committerDid string) {
115
+
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "Push"))
116
+
l.inner.Push(ctx, repo, ref, oldSha, newSha, committerDid)
117
+
}
118
+
119
+
func (l *loggingNotifier) Clone(ctx context.Context, repo *models.Repo) {
120
+
ctx = tlog.IntoContext(ctx, tlog.SubLogger(l.logger, "Clone"))
121
+
l.inner.Clone(ctx, repo)
122
+
}
+224
appview/notify/webhook/notifier.go
+224
appview/notify/webhook/notifier.go
···
1
+
package webhook
2
+
3
+
import (
4
+
"bytes"
5
+
"context"
6
+
"crypto/hmac"
7
+
"crypto/sha256"
8
+
"encoding/hex"
9
+
"encoding/json"
10
+
"fmt"
11
+
"io"
12
+
"log/slog"
13
+
"net/http"
14
+
"time"
15
+
16
+
"github.com/avast/retry-go/v4"
17
+
"github.com/google/uuid"
18
+
"tangled.org/core/appview/db"
19
+
"tangled.org/core/appview/models"
20
+
"tangled.org/core/appview/notify"
21
+
"tangled.org/core/log"
22
+
)
23
+
24
+
type Notifier struct {
25
+
notify.BaseNotifier
26
+
db *db.DB
27
+
logger *slog.Logger
28
+
client *http.Client
29
+
}
30
+
31
+
func NewNotifier(database *db.DB) *Notifier {
32
+
return &Notifier{
33
+
db: database,
34
+
logger: log.New("webhook-notifier"),
35
+
client: &http.Client{
36
+
Timeout: 30 * time.Second,
37
+
},
38
+
}
39
+
}
40
+
41
+
var _ notify.Notifier = &Notifier{}
42
+
43
+
func (w *Notifier) Push(ctx context.Context, repo *models.Repo, ref, oldSha, newSha, committerDid string) {
44
+
webhooks, err := db.GetActiveWebhooksForRepo(w.db, repo.RepoAt())
45
+
if err != nil {
46
+
w.logger.Error("failed to get webhooks for repo", "repo", repo.RepoAt(), "err", err)
47
+
return
48
+
}
49
+
50
+
var pushWebhooks []models.Webhook
51
+
for _, webhook := range webhooks {
52
+
if webhook.HasEvent(models.WebhookEventPush) {
53
+
pushWebhooks = append(pushWebhooks, webhook)
54
+
}
55
+
}
56
+
57
+
if len(pushWebhooks) == 0 {
58
+
return
59
+
}
60
+
61
+
payload, err := w.buildPushPayload(repo, ref, oldSha, newSha, committerDid)
62
+
if err != nil {
63
+
w.logger.Error("failed to build push payload", "repo", repo.RepoAt(), "err", err)
64
+
return
65
+
}
66
+
67
+
for _, webhook := range pushWebhooks {
68
+
go w.sendWebhook(ctx, webhook, string(models.WebhookEventPush), payload)
69
+
}
70
+
}
71
+
72
+
func (w *Notifier) buildPushPayload(repo *models.Repo, ref, oldSha, newSha, committerDid string) (*models.WebhookPayload, error) {
73
+
owner := repo.Did
74
+
75
+
pusher := committerDid
76
+
if committerDid == "" {
77
+
pusher = owner
78
+
}
79
+
80
+
repository := models.WebhookRepository{
81
+
Name: repo.Name,
82
+
FullName: fmt.Sprintf("%s/%s", repo.Did, repo.Name),
83
+
Description: repo.Description,
84
+
Fork: repo.Source != "",
85
+
HtmlUrl: fmt.Sprintf("https://%s/%s/%s", repo.Knot, repo.Did, repo.Name),
86
+
CloneUrl: fmt.Sprintf("https://%s/%s/%s", repo.Knot, repo.Did, repo.Name),
87
+
SshUrl: fmt.Sprintf("ssh://git@%s/%s/%s", repo.Knot, repo.Did, repo.Name),
88
+
CreatedAt: repo.Created.Format(time.RFC3339),
89
+
UpdatedAt: repo.Created.Format(time.RFC3339),
90
+
Owner: models.WebhookUser{
91
+
Did: owner,
92
+
},
93
+
}
94
+
95
+
if repo.Website != "" {
96
+
repository.Website = repo.Website
97
+
}
98
+
if repo.RepoStats != nil {
99
+
repository.StarsCount = repo.RepoStats.StarCount
100
+
repository.OpenIssues = repo.RepoStats.IssueCount.Open
101
+
}
102
+
103
+
payload := &models.WebhookPayload{
104
+
Ref: ref,
105
+
Before: oldSha,
106
+
After: newSha,
107
+
Repository: repository,
108
+
Pusher: models.WebhookUser{
109
+
Did: pusher,
110
+
},
111
+
}
112
+
113
+
return payload, nil
114
+
}
115
+
116
+
func (w *Notifier) sendWebhook(ctx context.Context, webhook models.Webhook, event string, payload *models.WebhookPayload) {
117
+
deliveryId := uuid.New().String()
118
+
119
+
payloadBytes, err := json.Marshal(payload)
120
+
if err != nil {
121
+
w.logger.Error("failed to marshal webhook payload", "webhook_id", webhook.Id, "err", err)
122
+
return
123
+
}
124
+
125
+
req, err := http.NewRequestWithContext(ctx, "POST", webhook.Url, bytes.NewReader(payloadBytes))
126
+
if err != nil {
127
+
w.logger.Error("failed to create webhook request", "webhook_id", webhook.Id, "err", err)
128
+
return
129
+
}
130
+
131
+
shortSha := payload.After[:7]
132
+
133
+
req.Header.Set("Content-Type", "application/json")
134
+
req.Header.Set("User-Agent", "Tangled-Hook/"+shortSha)
135
+
req.Header.Set("X-Tangled-Event", event)
136
+
req.Header.Set("X-Tangled-Hook-ID", fmt.Sprintf("%d", webhook.Id))
137
+
req.Header.Set("X-Tangled-Delivery", deliveryId)
138
+
req.Header.Set("X-Tangled-Repo", payload.Repository.FullName)
139
+
140
+
if webhook.Secret != "" {
141
+
signature := w.computeSignature(payloadBytes, webhook.Secret)
142
+
req.Header.Set("X-Tangled-Signature-256", "sha256="+signature)
143
+
}
144
+
145
+
delivery := &models.WebhookDelivery{
146
+
WebhookId: webhook.Id,
147
+
Event: event,
148
+
DeliveryId: deliveryId,
149
+
Url: webhook.Url,
150
+
RequestBody: string(payloadBytes),
151
+
}
152
+
153
+
retryOpts := []retry.Option{
154
+
retry.Attempts(3),
155
+
retry.Delay(1 * time.Second),
156
+
retry.MaxDelay(10 * time.Second),
157
+
retry.DelayType(retry.BackOffDelay),
158
+
retry.LastErrorOnly(true),
159
+
retry.OnRetry(func(n uint, err error) {
160
+
w.logger.Info("retrying webhook delivery",
161
+
"webhook_id", webhook.Id,
162
+
"attempt", n+1,
163
+
"err", err)
164
+
}),
165
+
retry.Context(ctx),
166
+
retry.RetryIf(func(err error) bool {
167
+
return err != nil
168
+
}),
169
+
}
170
+
171
+
var resp *http.Response
172
+
err = retry.Do(func() error {
173
+
var err error
174
+
resp, err = w.client.Do(req)
175
+
if err != nil {
176
+
return err
177
+
}
178
+
if resp.StatusCode >= 500 {
179
+
defer resp.Body.Close()
180
+
return fmt.Errorf("server error: %d", resp.StatusCode)
181
+
}
182
+
return nil
183
+
}, retryOpts...)
184
+
185
+
if err != nil {
186
+
w.logger.Error("webhook request failed after retries", "webhook_id", webhook.Id, "err", err)
187
+
delivery.Success = false
188
+
delivery.ResponseBody = err.Error()
189
+
} else {
190
+
defer resp.Body.Close()
191
+
192
+
delivery.ResponseCode = resp.StatusCode
193
+
delivery.Success = resp.StatusCode >= 200 && resp.StatusCode < 300
194
+
195
+
bodyBytes, err := io.ReadAll(io.LimitReader(resp.Body, 10*1024))
196
+
if err != nil {
197
+
w.logger.Warn("failed to read webhook response body", "webhook_id", webhook.Id, "err", err)
198
+
} else {
199
+
delivery.ResponseBody = string(bodyBytes)
200
+
}
201
+
202
+
if !delivery.Success {
203
+
w.logger.Warn("webhook delivery failed",
204
+
"webhook_id", webhook.Id,
205
+
"status", resp.StatusCode,
206
+
"url", webhook.Url)
207
+
} else {
208
+
w.logger.Info("webhook delivered successfully",
209
+
"webhook_id", webhook.Id,
210
+
"url", webhook.Url,
211
+
"delivery_id", deliveryId)
212
+
}
213
+
}
214
+
215
+
if err := db.AddWebhookDelivery(w.db, delivery); err != nil {
216
+
w.logger.Error("failed to record webhook delivery", "webhook_id", webhook.Id, "err", err)
217
+
}
218
+
}
219
+
220
+
func (w *Notifier) computeSignature(payload []byte, secret string) string {
221
+
mac := hmac.New(sha256.New, []byte(secret))
222
+
mac.Write(payload)
223
+
return hex.EncodeToString(mac.Sum(nil))
224
+
}